diff --git a/tools/engine.py b/tools/engine.py index d743677..86f88fd 100644 --- a/tools/engine.py +++ b/tools/engine.py @@ -55,7 +55,7 @@ SYSTEM_PROMPT = Template("""You are the Dungeon Master for **The Chaos**, a solo - 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. +- Never present predefined choices — the player decides freely what to do. - Each turn should advance the story meaningfully. ## Game Rules (Quick Reference) @@ -88,7 +88,6 @@ 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, @@ -99,7 +98,6 @@ IMPORTANT: End every response with a JSON fenced code block: ``` 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. @@ -107,6 +105,29 @@ Rules for the JSON block: - **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 @@ -117,8 +138,11 @@ $character ### World $world -### Recent Events -$log""") +### Recent Log +$log + +### Recent Story (last turns from the book) +$story""") # trailing """ is intentional — the template ends here @@ -211,7 +235,10 @@ class GameEngine: 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) + story = self._read_recent_book() + return SYSTEM_PROMPT.substitute( + character=char, world=world, log=log, story=story + ) def build_user_message( self, @@ -224,19 +251,34 @@ class GameEngine: 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: - 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." - ) + 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. Then present new choices. " - "End with a JSON block." + "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) @@ -354,6 +396,242 @@ class GameEngine: 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 @@ -370,7 +648,7 @@ class GameEngine: err = "Unknown error" return GenerationResult(narrative="", error=err) - # Extract JSON block — find the last ```json ... ``` block + # Try to find a ```json ... ``` block json_pattern = r"```json\s*\n?(.*?)\n?```" matches = re.findall(json_pattern, text, re.DOTALL) @@ -379,14 +657,21 @@ class GameEngine: 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 + 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, diff --git a/tools/run.py b/tools/run.py index 58b8ec9..dd467e9 100755 --- a/tools/run.py +++ b/tools/run.py @@ -17,6 +17,7 @@ from pathlib import Path from textual import on from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical, VerticalScroll +from textual.screen import Screen from textual.widgets import Button, Input, Static, TabbedContent, TabPane from rich.markdown import Markdown as RichMarkdown from rich.theme import Theme @@ -241,6 +242,76 @@ class AmbiencePlayer: app_ambience_player = None +# ── Roll Modal ─────────────────────────────────────────── +class RollModal(Screen): + """Overlay asking the player to roll physical dice and enter the result.""" + + CSS = """ + RollModal { + align: center middle; + background: rgba(0, 0, 0, 0.75); + } + #roll-dialog { + width: 44; + height: auto; + padding: 2 3; + background: #2a2a3a; + border: thick #e0ad4c; + } + #roll-title { + text-style: bold; + color: #ffd93d; + text-align: center; + height: 3; + } + #roll-reason { + color: #c0b090; + text-align: center; + height: 3; + } + #roll-input { + margin: 1 0; + } + #roll-submit { + width: 100%; + } + #roll-hint { + color: #888888; + text-align: center; + height: 1; + } + """ + + def __init__(self, dice: str, reason: str) -> None: + super().__init__() + self.dice = dice + self.reason = reason + + def compose(self) -> ComposeResult: + with Vertical(id="roll-dialog"): + yield Static(f"[bold]🎲 ROLL {self.dice}[/bold]", id="roll-title") + yield Static(f"Reason: {self.reason}", id="roll-reason") + yield Input( + placeholder="Enter the number you rolled...", + id="roll-input", + ) + yield Button("Submit", id="roll-submit", variant="primary") + yield Static("(or press Enter)", id="roll-hint") + + def on_input_submitted(self, event: Input.Submitted) -> None: + self._submit(event.value) + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "roll-submit": + inp = self.query_one("#roll-input", Input) + self._submit(inp.value) + + def _submit(self, value: str) -> None: + val = value.strip() + if val: + self.dismiss(val) + + # ── Auto-refreshing panels ─────────────────────────────── class AutoStatic(Static): def load(self): @@ -358,17 +429,6 @@ class ChaosTUI(App): padding: 1 2; height: auto; } - #play-choices { - height: auto; - min-height: 3; - background: #1e1e2a; - padding: 0 1; - align: center middle; - } - #play-choices Button { - margin: 0 1; - min-width: 12; - } #play-status { background: #1a1a2a; color: #e0b060; @@ -480,6 +540,11 @@ class ChaosTUI(App): # Thinking animation self._thinking_dots = 0 self._thinking_timer_handle = None + self._dm_action = "DM is weaving the narrative" + + # Player roll state (thread-safe) + self._roll_event = threading.Event() + self._roll_result: str | None = None # Book viewer state self._book_page = 0 @@ -496,7 +561,6 @@ class ChaosTUI(App): with TabPane("PLAY", id="play-tab"): with VerticalScroll(id="play-scroll"): yield Static("*Awaiting the fates...*", id="play-narrative") - yield Horizontal(id="play-choices") yield Static("", id="play-status") yield Input( placeholder="Type your action and press Enter...", @@ -548,9 +612,8 @@ class ChaosTUI(App): input_widget = self.query_one("#play-input", Input) input_widget.disabled = True - input_widget.placeholder = "LLM is thinking..." + input_widget.placeholder = "DM is at work..." - self._clear_choices() self._show_thinking() # Run generation in a daemon thread so it doesn't block the UI @@ -562,23 +625,32 @@ class ChaosTUI(App): t.start() def _run_generation(self, player_action: str | None) -> None: - """Worker thread: calls engine.generate() and posts result back.""" - # Provide previous narrative as context on subsequent calls + """Worker thread: calls engine.generate_with_tools() and posts result back.""" last_narrative = self._last_narrative if self._last_narrative else None - result = self.engine.generate( + def on_thought(thought: str) -> None: + self.call_from_thread(self._on_thought, thought) + + def on_action(action: str) -> None: + self.call_from_thread(self._on_action, action) + + result = self.engine.generate_with_tools( player_action=player_action, last_narrative=last_narrative, + on_thought=on_thought, + on_action=on_action, + on_player_roll=self._on_player_roll, ) self.call_from_thread(self._on_generation_done, result, player_action) 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 status = self.query_one("#play-status", Static) status.add_class("processing") - status.update("✦ LLM is weaving the narrative ✦") + status.update(f"✦ {self._dm_action} ✦") self._thinking_timer_handle = self.set_interval( 0.5, self._tick_thinking ) @@ -592,14 +664,51 @@ class ChaosTUI(App): status.remove_class("processing") status.update("") + def _on_thought(self, thought: str) -> None: + """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 + status = self.query_one("#play-status", Static) + status.add_class("processing") + status.update(f"✦ {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 + status = self.query_one("#play-status", Static) + status.add_class("processing") + status.update(f"✦ {action} ✦") + + def _on_player_roll(self, dice: str, reason: str) -> str: + """Called from worker thread. Shows roll popup, blocks until player responds.""" + self.call_from_thread(self._show_roll_modal, dice, reason) + self._roll_event.wait() + self._roll_event.clear() + result = self._roll_result + self._roll_result = None + return result or "0" + + def _show_roll_modal(self, dice: str, reason: str) -> None: + """Push the RollModal screen (runs on main thread).""" + self._roll_event.clear() + self._roll_result = None + + def on_dismiss(value: str) -> None: + self._roll_result = value + self._roll_event.set() + + self.push_screen(RollModal(dice, reason), on_dismiss) + def _tick_thinking(self) -> None: - """Animate the thinking dots.""" + """Animate the thinking dots on the current DM action.""" if not self._is_processing: return self._thinking_dots = (self._thinking_dots + 1) % 4 dots = "." * self._thinking_dots status = self.query_one("#play-status", Static) - status.update(f"✦ LLM is weaving the narrative{dots} ✦") + status.update(f"✦ {self._dm_action}{dots} ✦") def _on_generation_done( self, result: GenerationResult, player_action: str | None @@ -636,7 +745,6 @@ class ChaosTUI(App): def _display_scene(self, result: GenerationResult) -> None: """Update the UI with a new scene.""" self._set_narrative(result.narrative) - self._set_choices(result.choices) self._enable_input() def _enable_input(self) -> None: @@ -653,17 +761,6 @@ class ChaosTUI(App): scroll = self.query_one("#play-scroll", VerticalScroll) scroll.scroll_home(animate=False) - def _clear_choices(self) -> None: - container = self.query_one("#play-choices", Horizontal) - container.remove_children() - - def _set_choices(self, choices: list[str]) -> None: - container = self.query_one("#play-choices", Horizontal) - container.remove_children() - for choice in choices: - btn = Button(choice, classes="choice-btn") - container.mount(btn) - def _show_error(self, error: str) -> None: self._set_narrative( f"**Error:** {error}\n\n" @@ -681,16 +778,8 @@ class ChaosTUI(App): event.stop() self._handle_player_action(action) - @on(Button.Pressed, ".choice-btn") - def on_choice_clicked(self, event: Button.Pressed) -> None: - """Player clicked a choice button.""" - if self._is_processing: - return - action = event.button.label - self._handle_player_action(str(action)) - def _handle_player_action(self, action: str) -> None: - """Common handler for player actions from input or buttons.""" + """Handle a player action typed in the input.""" # Log the action from datetime import datetime timestamp = datetime.now().strftime("%H:%M")