diff --git a/session/config.json b/session/config.json index e0df9e0..b095244 100644 --- a/session/config.json +++ b/session/config.json @@ -1,8 +1,8 @@ { "llm": { - "model": "openai/gpt-4o-mini", - "api_key": "not-needed", + "model": "openai/deepseek-r1", + "api_key": null, "api_base": "http://localhost:8080/v1", - "temperature": 0.8 + "temperature": 0.7 } } diff --git a/tools/engine.py b/tools/engine.py index 86f88fd..61a7658 100644 --- a/tools/engine.py +++ b/tools/engine.py @@ -35,6 +35,7 @@ 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 @@ -46,6 +47,20 @@ class GenerationResult: 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 + 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 + 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. @@ -83,50 +98,46 @@ Favourable +1, Risky -1, Desperate -2, Well-prepared +1, Poor visibility -1, Rel ### 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: +## How Turns Work -```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": [] -} -``` +Each turn follows this sequence: +1. The player's action or response is given to you. +2. Think about what happens. Read game state files, roll dice, or ask the player to roll. +3. When ready, call **finalize_turn** to complete the turn. -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. +The **finalize_turn** tool produces all data for this turn: +- **book_log** — Narrative of what happened this turn. Appended to the story book. +- **user_prompt** — What the player sees next: describe the situation and ask what they do. +- **ambience** — One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds. +- **character_updates** — Full character sheet (ONLY if HP/cash/gear/stats changed, otherwise omit). +- **world_updates** — Full world state (ONLY if NPCs/locations/threads changed, otherwise omit). +- **journal_add** — New TODO items. +- **journal_done** — Completed TODO items. + +IMPORTANT: You MUST call **finalize_turn** to end the turn. Until then you will be called again to continue thinking and gathering information. ## Available Tools -You may use these tools before writing your final JSON block. Tool calls go in their own fenced code 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: +You may also show reasoning inline: ```thought Your reasoning here ``` -Call tools as needed, then end with the final JSON block. +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. + +- **read_file** — Read a game state file. `{"file": "character|world|book|log|journal", "dm_status": "..."}` +- **roll** — Auto-roll dice (outcome shown in status). `{"dice": "2d6", "modifier": "-1", "dm_status": "..."}` +- **player_roll** — Ask the player to roll physical dice. **Use when the outcome is uncertain.** `{"dice": "2d6", "reason": "why", "dm_status": "..."}` +- **finalize_turn** — **Complete the turn.** Provide all turn data as args. **Must include** `"dm_status"`. 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. @@ -202,6 +213,15 @@ class GameEngine: 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: @@ -243,42 +263,46 @@ class GameEngine: def build_user_message( self, player_action: str | None = None, - last_narrative: 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_narrative: - parts.append(f"## Previously\n{last_narrative}") + + 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_narrative else True + ) if not last_prompt else True - if not player_action and not last_narrative: + if not player_action and not last_prompt: 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." + "Continue the story from where it left off. Think, " + "gather information, then call finalize_turn." ) 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." + "setting, then call finalize_turn." ) 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." + "mechanics where appropriate. Think, gather information, " + "then call finalize_turn to complete the turn." ) return "\n\n".join(parts) @@ -297,7 +321,7 @@ class GameEngine: """ system = self.build_system_prompt() user = self.build_user_message( - player_action=player_action, last_narrative=last_narrative + player_action=player_action, last_prompt=last_narrative ) messages = [ @@ -316,15 +340,7 @@ class GameEngine: ) # 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 + self._set_llm_env() try: response = litellm.completion( @@ -332,6 +348,7 @@ class GameEngine: messages=messages, temperature=self.temperature, stream=False, + timeout=30, ) text = response.choices[0].message.content or "" except Exception as e: @@ -353,7 +370,7 @@ class GameEngine: """ system = self.build_system_prompt() user = self.build_user_message( - player_action=player_action, last_narrative=last_narrative + player_action=player_action, last_prompt=last_narrative ) messages = [ @@ -369,14 +386,7 @@ class GameEngine: }) 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 + self._set_llm_env() try: response = litellm.completion( @@ -384,6 +394,7 @@ class GameEngine: messages=messages, temperature=self.temperature, stream=True, + timeout=30, ) full_text = "" for chunk in response: @@ -415,8 +426,24 @@ class GameEngine: "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)"}, }, + "finalize_turn": { + "description": "Complete the turn with all required data.", + "args": { + "book_log": "Narrative of what happened (appended to story book)", + "user_prompt": "What the player sees next — describe and ask what they do", + "ambience": "Optional: soundscape name", + "character_updates": "Optional: full character sheet if changed", + "world_updates": "Optional: full world state if changed", + "journal_add": "Optional: list of new TODO items", + "journal_done": "Optional: list of completed TODO items", + }, + }, } + 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 = { @@ -454,7 +481,12 @@ class GameEngine: @staticmethod def _describe_tool_action(tool_name: str, args: dict) -> str: - """Return a user-facing status message for a tool call.""" + """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", @@ -474,8 +506,6 @@ class GameEngine: 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}..." @@ -484,6 +514,7 @@ class GameEngine: fn_map = { "read_file": self._tool_read_file, "roll": self._tool_roll, + "think": self._tool_think, } fn = fn_map.get(tool_name) if not fn: @@ -526,26 +557,25 @@ class GameEngine: def generate_with_tools( self, player_action: str | None = None, - last_narrative: str | None = None, + last_prompt: str | None = None, on_thought: callable = None, on_action: callable = None, on_player_roll: callable = None, - ) -> GenerationResult: + ) -> TurnResult: """ 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. + 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` is called for each thought block (may be from a - worker thread — use call_from_thread in the TUI). + `on_thought` / `on_action` 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_narrative=last_narrative, + last_prompt=last_prompt, ) messages: list[dict] = [ @@ -553,47 +583,102 @@ class GameEngine: {"role": "user", "content": user}, ] + self._set_llm_env() + 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 + return TurnResult(error="litellm not installed") max_rounds = 10 + debug_entries: list[str] = [] for round_idx in range(max_rounds): + round_log: list[str] = [f"── Round {round_idx + 1} ──"] + try: response = litellm.completion( model=self.model, messages=messages, temperature=self.temperature, stream=False, + timeout=30, ) text = response.choices[0].message.content or "" except Exception as e: - return GenerationResult(narrative="", error=f"LLM call failed: {e}") + return TurnResult(error=f"LLM call failed: {e}") + # 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()) + # Tool calls tool_calls = self._extract_tool_calls(text) - final_data = self._extract_final_json(text) + 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)}") + + # finalize_turn present → validate and return + if finalize_call: + args = finalize_call.get("args", {}) + errs = [] + if not args.get("dm_status"): + errs.append("dm_status is required") + if not args.get("book_log"): + errs.append("book_log is required") + if not args.get("user_prompt"): + errs.append("user_prompt is required") + if errs: + round_log.append(f" finalize_turn validation errors: {', '.join(errs)}") + debug_entries.append("\n".join(round_log)) + messages.append({"role": "assistant", "content": text}) + messages.append({ + "role": "user", + "content": f"## Validation Error\nfinalize_turn missing: {', '.join(errs)}. Please provide all required fields and call finalize_turn again." + }) + continue + return TurnResult( + book_log=args.get("book_log", ""), + user_prompt=args.get("user_prompt", ""), + ambience=args.get("ambience"), + character_updates=args.get("character_updates"), + world_updates=args.get("world_updates"), + journal_add=args.get("journal_add", []), + journal_done=args.get("journal_done", []), + ) + + # Execute other tools + if other_calls: results = [] - for tc in tool_calls: + 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"Describe what the DM is doing (e.g. " + f'`"dm_status": "consulting the archives"`). Please retry.' + ) + results.append(err_msg) + round_log.append(f" {name}: MISSING dm_status") + continue + if on_action: on_action(self._describe_tool_action(name, args)) if name == "player_roll" and on_player_roll: @@ -604,32 +689,28 @@ class GameEngine: 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") messages.append({"role": "assistant", "content": text}) messages.append({ "role": "user", "content": "## Tool Results\n\n" + "\n\n".join(results), }) - continue # Another round + debug_entries.append("\n".join(round_log)) + continue - # 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", []), - ) + # No tools, no finalize → remind LLM + round_log.append(" no tool calls — prompted to use tools") + debug_entries.append("\n".join(round_log)) + messages.append({"role": "assistant", "content": text}) + messages.append({ + "role": "user", + "content": "## Instructions\nUse tools to gather information or call **finalize_turn** to complete the turn." + }) - # Fallback: no tool calls, no JSON → normal parse - return self.parse_response(text) - - return GenerationResult( - narrative="", - error="Tool loop exceeded max rounds (10)", + debug_text = "\n\n".join(debug_entries) + 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 ──────────────────────────────────────────────── @@ -686,8 +767,8 @@ class GameEngine: # ── State Persistence ─────────────────────────────────────────────── - def apply_state(self, result: GenerationResult) -> None: - """Write state changes from a GenerationResult to disk.""" + def apply_state(self, result: TurnResult) -> None: + """Write state changes from a TurnResult to disk.""" if result.character_updates: CHAR_PATH.write_text(result.character_updates.strip() + "\n") @@ -695,9 +776,6 @@ class GameEngine: 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") diff --git a/tools/run.py b/tools/run.py index dd467e9..c03f9cb 100755 --- a/tools/run.py +++ b/tools/run.py @@ -23,7 +23,7 @@ from rich.markdown import Markdown as RichMarkdown from rich.theme import Theme # ── Game engine ───────────────────────────────────────── -from engine import GameEngine, GenerationResult +from engine import GameEngine, GenerationResult, TurnResult # ── Optional miniaudio ──────────────────────────────────── try: @@ -533,12 +533,13 @@ class ChaosTUI(App): self.engine = GameEngine() # Game loop state - self._last_narrative: str = "" - self._last_result: GenerationResult | None = None + self._last_prompt: str = "" + self._last_result: TurnResult | None = None self._is_processing: bool = False # Thinking animation - self._thinking_dots = 0 + self._spinner_frames = ["◴", "◷", "◶", "◵"] + self._thinking_frame = 0 self._thinking_timer_handle = None self._dm_action = "DM is weaving the narrative" @@ -605,7 +606,7 @@ class ChaosTUI(App): # ── Game Loop ───────────────────────────────────────── def _call_llm(self, player_action: str | None = None): - """Called when we need new content from the LLM (scene or resolution).""" + """Called when the player has acted — sends their action to the LLM.""" if self._is_processing: return self._is_processing = True @@ -626,7 +627,7 @@ class ChaosTUI(App): def _run_generation(self, player_action: str | None) -> None: """Worker thread: calls engine.generate_with_tools() and posts result back.""" - last_narrative = self._last_narrative if self._last_narrative else None + last_prompt = self._last_prompt if self._last_prompt else None def on_thought(thought: str) -> None: self.call_from_thread(self._on_thought, thought) @@ -636,7 +637,7 @@ class ChaosTUI(App): result = self.engine.generate_with_tools( player_action=player_action, - last_narrative=last_narrative, + last_prompt=last_prompt, on_thought=on_thought, on_action=on_action, on_player_roll=self._on_player_roll, @@ -647,10 +648,11 @@ class ChaosTUI(App): def _show_thinking(self) -> None: """Show the thinking indicator and start the animation timer.""" self._dm_action = "DM is weaving the narrative" - self._thinking_dots = 0 + self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") - status.update(f"✦ {self._dm_action} ✦") + spinner = self._spinner_frames[0] + status.update(f"✦ {spinner} {self._dm_action} ✦") self._thinking_timer_handle = self.set_interval( 0.5, self._tick_thinking ) @@ -668,18 +670,20 @@ class ChaosTUI(App): """Display a thought from the DM in the status bar.""" display = thought[:60] + "…" if len(thought) > 60 else thought self._dm_action = display - self._thinking_dots = 0 + self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") - status.update(f"✦ {display} ✦") + spinner = self._spinner_frames[0] + status.update(f"✦ {spinner} {display} ✦") def _on_action(self, action: str) -> None: """Display a DM action (tool call) in the status bar.""" self._dm_action = action - self._thinking_dots = 0 + self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") - status.update(f"✦ {action} ✦") + spinner = self._spinner_frames[0] + status.update(f"✦ {spinner} {action} ✦") def _on_player_roll(self, dice: str, reason: str) -> str: """Called from worker thread. Shows roll popup, blocks until player responds.""" @@ -702,49 +706,42 @@ class ChaosTUI(App): self.push_screen(RollModal(dice, reason), on_dismiss) def _tick_thinking(self) -> None: - """Animate the thinking dots on the current DM action.""" + """Animate the spinner on the current DM action.""" if not self._is_processing: return - self._thinking_dots = (self._thinking_dots + 1) % 4 - dots = "." * self._thinking_dots + self._thinking_frame = (self._thinking_frame + 1) % len(self._spinner_frames) + spinner = self._spinner_frames[self._thinking_frame] status = self.query_one("#play-status", Static) - status.update(f"✦ {self._dm_action}{dots} ✦") + status.update(f"✦ {spinner} {self._dm_action} ✦") def _on_generation_done( - self, result: GenerationResult, player_action: str | None + self, result: TurnResult, player_action: str | None ) -> None: - """Handle the completed generation on the main thread.""" + """Handle the completed turn on the main thread.""" self._is_processing = False self._hide_thinking() if result.error: - self._show_error(result.error) + self._show_error(result.error, result.debug_info) return - # If this was a resolution (player acted), archive the previous turn - if player_action and self._last_narrative: - archive_text = ( - f"{self._last_narrative}\n\n" - f"---\n\n" - f"**Player chose:** {player_action}\n\n" - f"{result.narrative}" - ) - self.engine.archive_turn(archive_text) + # Archive the turn's book log + if result.book_log: + self.engine.archive_turn(result.book_log) # Apply state changes - if result.character_updates or result.world_updates: - self.engine.apply_state(result) + self.engine.apply_state(result) - # Display the scene + # Display the next user prompt self._display_scene(result) # Store for next turn - self._last_narrative = result.narrative + self._last_prompt = result.user_prompt self._last_result = result - def _display_scene(self, result: GenerationResult) -> None: - """Update the UI with a new scene.""" - self._set_narrative(result.narrative) + def _display_scene(self, result: TurnResult) -> None: + """Update the UI with the next user prompt.""" + self._set_narrative(result.user_prompt) self._enable_input() def _enable_input(self) -> None: @@ -761,11 +758,12 @@ class ChaosTUI(App): scroll = self.query_one("#play-scroll", VerticalScroll) scroll.scroll_home(animate=False) - def _show_error(self, error: str) -> None: - self._set_narrative( - f"**Error:** {error}\n\n" - "Check your session/config.json and ensure your LLM provider is running." - ) + def _show_error(self, error: str, debug_info: str = "") -> None: + text = f"**Error:** {error}\n\n" + if debug_info: + text += f"**Debug Info:**\n\n{debug_info}\n\n" + text += "Check your session/config.json and ensure your LLM provider is running." + self._set_narrative(text) self._enable_input() # ── Input handling ────────────────────────────────────