From 545d3bcac0ce38d4e6789510aff65332e7221a7a Mon Sep 17 00:00:00 2001 From: Dejvino Date: Tue, 30 Jun 2026 20:03:53 +0200 Subject: [PATCH] More refactors --- pre-commit.sh | 42 +- tools/engine.py | 908 ++--------------- tools/engine_lib/__init__.py | 0 tools/engine_lib/config.py | 74 ++ tools/engine_lib/context.py | 74 ++ tools/{ => engine_lib}/llm.py | 2 +- tools/{ => engine_lib}/models.py | 0 tools/engine_lib/parsing.py | 121 +++ .../{config_paths.py => engine_lib/paths.py} | 4 +- tools/{ => engine_lib}/prompts.py | 0 tools/{ => engine_lib}/state.py | 4 +- tools/engine_lib/strategies.py | 657 ++++++++++++ tools/{ => engine_lib}/tools_handler.py | 4 +- tools/engine_lib/validation.py | 74 ++ tools/run.py | 942 +++--------------- tools/run_ambience.py | 97 ++ tools/run_utils.py | 125 +++ tools/run_widgets.py | 157 +++ tools/test_imports.py | 17 +- tools/test_runtime.py | 17 +- 20 files changed, 1633 insertions(+), 1686 deletions(-) create mode 100644 tools/engine_lib/__init__.py create mode 100644 tools/engine_lib/config.py create mode 100644 tools/engine_lib/context.py rename tools/{ => engine_lib}/llm.py (98%) rename tools/{ => engine_lib}/models.py (100%) create mode 100644 tools/engine_lib/parsing.py rename tools/{config_paths.py => engine_lib/paths.py} (85%) rename tools/{ => engine_lib}/prompts.py (100%) rename tools/{ => engine_lib}/state.py (99%) create mode 100644 tools/engine_lib/strategies.py rename tools/{ => engine_lib}/tools_handler.py (99%) create mode 100644 tools/engine_lib/validation.py create mode 100644 tools/run_ambience.py create mode 100644 tools/run_utils.py create mode 100644 tools/run_widgets.py diff --git a/pre-commit.sh b/pre-commit.sh index 8ea292f..4ee37ce 100755 --- a/pre-commit.sh +++ b/pre-commit.sh @@ -1,15 +1,33 @@ #!/bin/bash -ERRORS=$(python3 -c "import os; [f for f in os.listdir('./tools') if f.endswith('.py') and os.path.getsize(os.path.join('./tools', f)) > 2048]") -if [ -z "$ERRORS" ]; then - echo "Compiling tools/*.py..." - if python3 -m compileall tools/*.py; then - echo "OK" - else - echo "Compilation failed" - exit 1 - fi -else - echo "You need to refactor this:" - echo "$ERRORS" +THRESHOLD=30000 # bytes — flag files over ~30 KB + +# Check all .py files in tools/ and tools/engine_lib/ +OVERSIZED=$(python3 -c " +import os +threshold = $THRESHOLD +dirs = ['./tools', './tools/engine_lib'] +for d in dirs: + if not os.path.isdir(d): + continue + for f in sorted(os.listdir(d)): + if not f.endswith('.py'): + continue + path = os.path.join(d, f) + size = os.path.getsize(path) + if size > threshold: + print(f'{path} ({size} bytes)') +") + +if [ -n "$OVERSIZED" ]; then + echo "Oversized files (>${THRESHOLD} bytes) — consider refactoring:" + echo "$OVERSIZED" + exit 1 +fi + +echo "Compiling tools/*.py and tools/engine_lib/*.py..." +if python3 -m compileall tools/*.py tools/engine_lib/*.py; then + echo "OK" +else + echo "Compilation failed" exit 1 fi diff --git a/tools/engine.py b/tools/engine.py index 6902aba..f00e9c7 100644 --- a/tools/engine.py +++ b/tools/engine.py @@ -2,249 +2,86 @@ """ 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. - -Split into sub-modules: paths, models, prompts, state, tools_handler, llm. +Thin coordinator that owns the GameEngine class. All heavy lifting is +delegated to sub-modules: paths, models, prompts, config, context, +state, tools_handler, llm, validation, parsing, strategies. """ from __future__ import annotations -import json -import re import sys -from collections import Counter -from datetime import datetime from pathlib import Path -from typing import Iterator, Optional -from paths import ( - CHAR_PATH, WORLD_PATH, BOOK_PATH, CONFIG_PATH, LOG_DIR, -) -from models import GenerationResult, TurnResult -from prompts import SYSTEM_PROMPT, PROSE_PROMPT -import state # read_file, read_recent_log, read_recent_book, truncate_world, append_llm_log -from tools_handler import ( - execute_tool, describe_tool_action, describe_change, - parse_changes_block, extract_tool_calls, -) -from llm import set_llm_env, call_llm +from engine_lib.paths import CONFIG_PATH +from engine_lib.models import GenerationResult, TurnResult +from engine_lib import config +from engine_lib import strategies -# ── Game Engine ──────────────────────────────────────────────────────────── class GameEngine: - """Owns the LLM interaction and game state persistence.""" + """Owns configuration and delegates generation to standalone strategies.""" def __init__(self, session_dir: str | Path | None = None): - from paths import SESSION_DIR + from engine_lib.paths import SESSION_DIR self.session_dir = Path(session_dir) if session_dir else SESSION_DIR - self.config: dict = {} - self._load_config() + self.config = config.load_config(CONFIG_PATH) - # ── 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, - "max_tokens": 300, - } - } - self._save_config() - else: - raw = CONFIG_PATH.read_text() - self.config = json.loads(raw) - 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") + # ── Config accessors ──────────────────────────────────────────────── @property def model(self) -> str: - return self.config.get("llm", {}).get("model", "ollama/llama3.1") + return config.get_model(self.config) @property def api_key(self) -> str | None: - return self.config.get("llm", {}).get("api_key") + return config.get_api_key(self.config) @property def api_base(self) -> str | None: - return self.config.get("llm", {}).get("api_base") + return config.get_api_base(self.config) @property def temperature(self) -> float: - return self.config.get("llm", {}).get("temperature", 0.8) + return config.get_temperature(self.config) @property def max_tokens(self) -> int: - return self.config.get("llm", {}).get("max_tokens", 512) + return config.get_max_tokens(self.config) @property def timeout(self) -> int: - return self.config.get("llm", {}).get("timeout", 120) + return config.get_timeout(self.config) - # ── Context Assembly ──────────────────────────────────────────────── - - def build_system_prompt(self) -> str: - """Assemble the system prompt with current game state.""" - char = state.read_file(CHAR_PATH) or "*No character sheet.*" - world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*" - log = state.read_recent_log() - story = state.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's Request\n{player_action}") - - has_existing_story = bool( - state.read_file(BOOK_PATH).strip() - ) if not last_prompt else True - - if not player_action and not last_prompt: - if has_existing_story: - raise RuntimeError(f"User action is required for every turn.") - else: - parts.append( - "## Instructions\n" - "This is a new story. Welcome the player and guide them through the game setup." - ) - else: - parts.append( - "## Instructions\n" - "Advance the story based on the player's request. " - "All state is shown above — write the outcome directly." - ) - return "\n\n".join(parts) - - # ── LLM Call ──────────────────────────────────────────────────────── + # ── Generation (delegated) ────────────────────────────────────────── 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. - """ - system = self.build_system_prompt() - user = self.build_user_message( - player_action=player_action, last_prompt=last_narrative + return strategies.generate( + player_action=player_action, + last_narrative=last_narrative, + model=self.model, + temperature=self.temperature, + timeout=self.timeout, + max_tokens=self.max_tokens, + api_key=self.api_key, + api_base=self.api_base, ) - 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_llm_env(self.model, self.api_key, self.api_base) - - try: - response = litellm.completion( - model=self.model, - messages=messages, - temperature=self.temperature, - stream=False, - timeout=self.timeout, - ) - 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 + def generate_stream(self, player_action=None, last_narrative=None): + yield from strategies.generate_stream( + player_action=player_action, + last_narrative=last_narrative, + model=self.model, + temperature=self.temperature, + timeout=self.timeout, + max_tokens=self.max_tokens, + api_key=self.api_key, + api_base=self.api_base, ) - 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 - - set_llm_env(self.model, self.api_key, self.api_base) - - try: - response = litellm.completion( - model=self.model, - messages=messages, - temperature=self.temperature, - stream=True, - timeout=self.timeout, - ) - full_text = "" - for chunk in response: - delta = chunk.choices[0].delta.content or "" - if delta: - full_text += delta - yield full_text - yield full_text - except Exception as e: - yield json.dumps({"error": f"LLM call failed: {e}"}) - - # ──── Three-phase generation ──────────────────────────────────────── - def generate_with_tools( self, player_action: str | None = None, @@ -254,301 +91,20 @@ class GameEngine: on_player_roll: callable = None, on_debug: callable = None, ) -> TurnResult: - """ - Three-phase generation: - - 1. **Prose** — LLM writes the full book_log from context + player action. - 2. **Summarize** — LLM condenses the book_log into one log line. - 3. **Extract** — LLM reads the book_log and outputs tool calls for state changes. - """ - set_llm_env(self.model, self.api_key, self.api_base) - import random - datetime_now = datetime.now() - state.append_llm_log(f"\n{'='*60}") - state.append_llm_log(f"=== Turn — {datetime_now.strftime('%Y-%m-%d %H:%M:%S')} ===") - state.append_llm_log(f"{'='*60}") - if player_action: - state.append_llm_log(f"Player: {player_action}") - elif last_prompt: - state.append_llm_log(f"Resume from: {last_prompt[:120]}") - - die_roll = random.randint(1, 6) - state.append_llm_log(f"Dice: {die_roll} (1d6)") - - book_log = None - changes_block = "" - log_entry = None - user_prompt = self._auto_prompt("") - ambience = None - debug_info = "" - changes = [] - - for outer_attempt in range(3): - # ── Phase 1: Prose ──────────────────────────────────────────── - if on_action: - on_action(f"Phase 1/3: writing story (dice={die_roll})") - if on_debug: - on_debug("phase", {"phase": 1, "name": "prose", "status": "start", "dice": die_roll, "outer_attempt": outer_attempt + 1}) - - system = PROSE_PROMPT.substitute( - character=state.read_file(CHAR_PATH) or "*No character sheet.*", - world=state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*", - log=state.read_recent_log(), - story=state.read_recent_book(), - ) - user = self.build_user_message( - player_action=player_action, - last_prompt=last_prompt, - ) - user += f"\n\n*A die is cast: **{die_roll}** (1d6).*" - - text = call_llm([ - {"role": "system", "content": system}, - {"role": "user", "content": user}, - ], model=self.model, temperature=self.temperature, timeout=self.timeout, - max_tokens=1024, label=f"Prose attempt {outer_attempt + 1}", on_debug=on_debug) - - if not text or not text.strip(): - if on_debug: - on_debug("phase", {"phase": 1, "status": "empty", "attempt": outer_attempt + 1}) - continue - - raw = text.strip() - changes_block = "" - if "### Changes" in raw: - parts = raw.split("### Changes", 1) - book_log = parts[0].strip() - changes_block = "### Changes" + parts[1] - else: - book_log = raw - if on_debug: - preview = book_log[:150].replace("\n", "\\n") - on_debug("phase", {"phase": 1, "status": "done", "chars": len(book_log), "changes": bool(changes_block), "preview": preview}) - - # ── Validation ──────────────────────────────────────────────── - if on_debug: - on_debug("phase", {"phase": 1, "name": "validation", "status": "start"}) - valid, reason = self._validate_narrative(book_log, on_debug=on_debug) - if not valid: - if on_debug: - on_debug("phase", {"phase": 1, "status": "validation_failed", "reason": reason, "outer_attempt": outer_attempt + 1}) - book_log = None - continue - - # ── Phase 2: Summarize ──────────────────────────────────────── - if on_action: - on_action("Phase 2/3: summarizing story") - if on_debug: - on_debug("phase", {"phase": 2, "name": "summarize", "status": "start"}) - - log_context = state.read_recent_log() - log_entry = None - for p2_attempt in range(2): - context = book_log - if changes_block: - context += f"\n\n{changes_block}" - text = call_llm([ - {"role": "user", "content": - f"Given the session log so far, summarize the new story in one line. " - f"Focus on who was involved (character and NPC names):\n\n" - f"## Session Log\n{log_context}\n\n" - f"## New Story\n{context}"} - ], model=self.model, temperature=self.temperature, timeout=self.timeout, - max_tokens=self.max_tokens, label=f"Summarize attempt {p2_attempt + 1}", on_debug=on_debug) - if text and text.strip(): - log_entry = text.strip().split("\n")[0][:300] - if on_debug: - on_debug("phase", {"phase": 2, "status": "done", "summary": log_entry}) - break - - if not log_entry: - log_entry = book_log.split("\n")[0][:120] - if on_debug: - on_debug("phase", {"phase": 2, "status": "fallback", "summary": log_entry}) - - # ── Phase 3: Extract state changes ──────────────────────────── - if on_action: - on_action("Phase 3/3: extracting state changes") - if on_debug: - on_debug("phase", {"phase": 3, "name": "extract", "status": "start"}) - - user_prompt = self._auto_prompt(book_log) - ambience = None - phase3_errors = [] - changes = [] - - # Step 1: Parse ### Changes block directly - if changes_block.strip(): - for tc in parse_changes_block(changes_block): - name = tc["tool"] - args = tc.get("args", {}) - if name == "finalize_turn": - continue - result = execute_tool(name, args) - if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"): - phase3_errors.append(f"{name}: {result}") - else: - desc = describe_change(name, args) - if desc: - changes.append(desc) - - # Step 2: LLM Phase 3 for finalize_turn + any extra changes - previous_attempt = None - phase3_ok = False - for p3_attempt in range(5): - current_char = state.read_file(CHAR_PATH) or "*No character.*" - current_world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world.*" - - phase3_prompt = ( - f"## Current Character\n{current_char}\n\n" - f"## Current World\n{current_world}\n\n" - f"## Story\n{book_log}\n\n" - ) - if changes_block.strip(): - phase3_prompt += ( - f"## Changes already applied\n{changes_block}\n\n" - f"Output the finalize_turn tool to end the turn. " - f"Add extra tool calls if you spot changes the list above missed.\n\n" - ) - else: - phase3_prompt += ( - f"Read the story and compare with current state. Output tool calls for any changes:\n\n" - ) - phase3_prompt += ( - f"Output ```tool blocks for changes only. Examples:\n\n" - ) - - if previous_attempt: - phase3_prompt += ( - f"--- PREVIOUS ATTEMPT (had errors) ---\n" - f"{previous_attempt['output']}\n\n" - f"--- FEEDBACK ---\n" - f"{previous_attempt['feedback']}\n\n" - f"Fix the issues above. Output corrected tool calls only.\n\n" - ) - - text = call_llm([ - {"role": "user", "content": phase3_prompt + - f"```tool\n{{\"tool\": \"modify_vitals\", \"args\": {{\"current_hp\": 5, \"cash\": 45}}}}\n```\n" - f"```tool\n{{\"tool\": \"modify_traits\", \"args\": {{\"dex\": 15}}}}\n```\n" - f"```tool\n{{\"tool\": \"add_to_inventory\", \"args\": {{\"item\": \"Silver key\"}}}}\n```\n" - f"```tool\n{{\"tool\": \"remove_from_inventory\", \"args\": {{\"item\": \"Torches (10)\"}}}}\n```\n" - f"```tool\n{{\"tool\": \"replace_gear\", \"args\": {{\"before\": \"Mace (1d6+1)\", \"after\": \"Mace (1d6+2, sharpened)\"}}}}\n```\n" - f"```tool\n{{\"tool\": \"add_note\", \"args\": {{\"note\": \"Found a hidden passage under the temple\"}}}}\n```\n" - f"```tool\n{{\"tool\": \"replace_note\", \"args\": {{\"before\": \"Old note text\", \"after\": \"New note text\"}}}}\n```\n" - f"```tool\n{{\"tool\": \"world_update\", \"args\": {{\"content\": \"# The World\\n\\n...full new world state...\"}}}}\n```\n" - f"```tool\n{{\"tool\": \"journal_update\", \"args\": {{\"add\": [\"Investigate the mine\"], \"done\": [\"Defeat the demon\"]}}}}\n```\n" - f"```tool\n{{\"tool\": \"finalize_turn\", \"args\": {{\"user_prompt\": \"What do you do?\", \"ambience\": \"dungeon\"}}}}\n```\n\n" - f"Only output tools for things that actually changed. Omit unchanged fields."} - ], model=self.model, temperature=self.temperature, timeout=self.timeout, - max_tokens=self.max_tokens, label=f"Extract attempt {p3_attempt + 1}", on_debug=on_debug) - - if not text or not text.strip(): - if on_debug: - on_debug("phase", {"phase": 3, "status": "empty", "attempt": p3_attempt + 1}) - continue - - tool_calls_list = extract_tool_calls( - text, round_num=p3_attempt + 1, on_debug=on_debug - ) - if on_debug and tool_calls_list: - names = [tc.get("tool", "?") for tc in tool_calls_list if tc.get("tool") != "finalize_turn"] - fin = any(tc.get("tool") == "finalize_turn" for tc in tool_calls_list) - on_debug("phase", {"phase": 3, "status": "tools_found", "tools": names, "has_finalize": fin}) - - errors = [] - attempt_changes = [] - for tc in tool_calls_list: - name = tc.get("tool", "?") - args = tc.get("args", {}) - if name == "finalize_turn": - if args.get("user_prompt"): - user_prompt = args["user_prompt"] - if args.get("ambience"): - ambience = args["ambience"] - continue - if on_action: - on_action(f"State: {describe_tool_action(name, args)}") - if on_debug: - on_debug("tool_call", {"round": p3_attempt + 1, "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 = execute_tool(name, args) - - if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"): - errors.append(f"{name}: {result}") - else: - desc = describe_change(name, args) - if desc: - attempt_changes.append(desc) - if on_debug: - on_debug("tool_result", {"round": p3_attempt + 1, "tool": name, "result": result}) - - if not errors: - phase3_ok = True - debug_info = "" - changes.extend(attempt_changes) - if on_debug: - on_debug("phase", {"phase": 3, "status": "done", "applied": len([tc for tc in tool_calls_list if tc.get("tool") != "finalize_turn"])}) - break - - phase3_errors = errors - debug_info = "; ".join(errors) - if on_debug: - on_debug("phase", {"phase": 3, "status": "errors", "errors": errors, "attempt": p3_attempt + 1}) - - feedback_lines = ["The previous tool calls had errors:"] - for e in errors: - feedback_lines.append(f"- {e}") - feedback_lines.append("") - feedback_lines.append("Fix ALL issues above. Use correct tool names, valid JSON, and reasonable values.") - previous_attempt = {"output": text, "feedback": "\n".join(feedback_lines)} - - if phase3_ok: - break - - if on_debug: - on_debug("phase", {"phase": 3, "status": "exhausted", "errors": phase3_errors}) - on_debug("phase", {"phase": 1, "status": "retry_after_phase3_failure", "outer_attempt": outer_attempt + 1}) - book_log = None - - if not book_log: - return TurnResult(error="Generation failed after exhausting all retries") - - # ── Finalize ────────────────────────────────────────────────────── - if on_action: - on_action("Turn complete") - if on_debug: - on_debug("phase_done", { - "book_log_chars": len(book_log), - "log_entry": log_entry, - "user_prompt": user_prompt, - "ambience": ambience, - "extract_errors": debug_info or None, - }) - - state.append_llm_log( - f"\n--- FINAL ---\n" - f"book_log: {book_log[:200]}\n" - f"log_entry: {log_entry}\n" - f"user_prompt: {user_prompt}\n" - f"ambience: {ambience}\n" + return strategies.generate_with_tools( + player_action=player_action, + last_prompt=last_prompt, + on_thought=on_thought, + on_action=on_action, + on_player_roll=on_player_roll, + on_debug=on_debug, + model=self.model, + temperature=self.temperature, + timeout=self.timeout, + max_tokens=self.max_tokens, + api_key=self.api_key, + api_base=self.api_base, ) - return TurnResult( - book_log=book_log, - log_entry=log_entry, - user_prompt=user_prompt, - ambience=ambience, - debug_info=debug_info, - changes=changes, - ) - - # ──── Single-call generation ───────────────────────────────────────── def generate_with_tools_single( self, @@ -559,368 +115,20 @@ class GameEngine: on_player_roll: callable = None, on_debug: callable = None, ) -> TurnResult: - """ - Single-call generation using tools. - - Uses a single LLM call with all tools available — LLM outputs - narrative + tool blocks in one go. No retry loop. - """ - datetime_now = datetime.now() - state.append_llm_log(f"\n{'='*60}") - state.append_llm_log(f"=== Turn — {datetime_now.strftime('%Y-%m-%d %H:%M:%S')} ===") - state.append_llm_log(f"{'='*60}") - if player_action: - state.append_llm_log(f"Player: {player_action}") - elif last_prompt: - state.append_llm_log(f"Resume from: {last_prompt[:120]}") - - strategy_name = "tools" - if on_action: - on_action(f"LLM: {self.model} | temp={self.temperature} | tokens={self.max_tokens} | strategy={strategy_name}") - if on_debug: - on_debug("config", {"model": self.model, "temperature": self.temperature, "max_tokens": self.max_tokens, "strategy": strategy_name}) - - import random - die_roll = random.randint(1, 6) - state.append_llm_log(f"Dice: {die_roll} (1d6)") - - system = """You are an RPG dungeon master. The player just took an action. - -Output ONLY ```tool blocks — no prose, no reasoning, no explanation outside tool blocks. Every piece of output must be in a tool block. - -Use these tools to perform every action. Wrap each in its own ```tool block: -```tool -{"tool": "narrative", "args": {"text": "The full vivid narrative prose goes here."}} -``` -```tool -{"tool": "modify_vitals", "args": {"current_hp": 5, "cash": 45}} -``` -```tool -{"tool": "modify_traits", "args": {"dex": 15}} -``` -```tool -{"tool": "add_to_inventory", "args": {"item": "Silver key"}} -``` -```tool -{"tool": "remove_from_inventory", "args": {"item": "Torches (10)"}} -``` -```tool -{"tool": "replace_gear", "args": {"before": "Mace (1d6+1)", "after": "Mace (1d6+2, sharpened)"}} -``` -```tool -{"tool": "add_note", "args": {"note": "Found a hidden passage under the temple"}} -``` -```tool -{"tool": "replace_note", "args": {"before": "Old note text", "after": "New note text"}} -``` -```tool -{"tool": "world_update", "args": {"content": "# The World\n\n...full new world state..."}} -``` -```tool -{"tool": "journal_update", "args": {"add": ["Investigate the mine"], "done": ["Defeat the demon"]}} -``` -```tool -{"tool": "finalize_turn", "args": {"user_prompt": "What do you do?", "ambience": "dungeon"}} -``` -""" - system += PROSE_PROMPT.substitute( - character=state.read_file(CHAR_PATH) or "*No character sheet.*", - world=state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*", - log=state.read_recent_log(), - story=state.read_recent_book(), - ) - - user = self.build_user_message( + return strategies.generate_with_tools_single( player_action=player_action, last_prompt=last_prompt, - ) - user += f"\n\n*A die is cast: **{die_roll}** (1d6).*" - - start_time = datetime.now() - set_llm_env(self.model, self.api_key, self.api_base) - state.append_llm_log(f"\n[TOOL] Single call — {len(system)} chars system, {len(user)} chars user") - state.append_llm_log(f"System preview: {system.split(chr(10))[0][:80]}...") - state.append_llm_log(f"User preview: {user.split(chr(10))[0][:80]}...") - - text = call_llm( - [{"role": "system", "content": system}, - {"role": "user", "content": user}], - model=self.model, temperature=self.temperature, timeout=self.timeout, - max_tokens=4096, label="Single tool call", on_debug=on_debug, - ) - - total_elapsed = (datetime.now() - start_time).total_seconds() * 1000 - if text: - state.append_llm_log(f"\n[TOOL] got {len(text)} chars in {total_elapsed:.1f}ms") - - if not text or not text.strip(): - return TurnResult(error="Single tool call returned empty response") - - raw = text.strip() - book_log = "" - log_entry = None - user_prompt = self._auto_prompt("") - ambience = None - tool_calls = [] - changes = [] - phase3_errors = [] - - tool_pattern = r"```tool\s*\n?(.*?)\n?```" - matches = re.findall(tool_pattern, text, re.DOTALL) - if matches: - for block in matches: - block = block.strip() - try: - tc = json.loads(block) - tool_calls.append(tc) - name = tc.get("tool", "unknown") - args = tc.get("args", {}) - state.append_llm_log(f"\n[EXTRACT] {name}: {json.dumps(args)[:100]}") - - if name == "narrative": - book_log = args.get("text", book_log) - elif name == "finalize_turn": - if args.get("user_prompt"): - user_prompt = args["user_prompt"] - if args.get("ambience"): - ambience = args["ambience"] - except json.JSONDecodeError as e: - state.append_llm_log(f"\n[EXTRACT] bad JSON: {e}") - continue - - log_entry = None - if book_log: - clean = re.sub(r'\s+', ' ', book_log).strip() - first_sentence = re.split(r'(?<=[.!?])\s+', clean) - if first_sentence: - log_entry = first_sentence[0].strip()[:200] - else: - log_entry = clean[:200] - state.append_llm_log(f"\n[SUMMARY] \"{log_entry}\"") - - extr_start = datetime.now() - for tc in tool_calls: - name = tc.get("tool", "unknown") - args = tc.get("args", {}) - if name in ("finalize_turn", "narrative"): - continue - result = execute_tool(name, args) - if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"): - phase3_errors.append(f"{name}: {result}") - else: - desc = describe_change(name, args) - if desc: - changes.append(desc) - - apply_elapsed = (datetime.now() - extr_start).total_seconds() * 1000 - state.append_llm_log(f"\n[APPLY] {len(changes)} changes in {apply_elapsed:.1f}ms") - else: - state.append_llm_log(f"\n[TOOL] no tool blocks found") - - elapsed = (datetime.now() - start_time).total_seconds() * 1000 - - if on_action: - on_action("Turn complete") - if on_debug: - applied = len([tc for tc in tool_calls if tc.get("tool") != "finalize_turn"]) - on_debug("phase_done", { - "book_log_chars": len(book_log), - "log_entry": log_entry, - "user_prompt": user_prompt, - "ambience": ambience, - "extract_errors": phase3_errors or None, - "total_elapsed_ms": elapsed, - "tool_calls_count": len(tool_calls), - "applied_changes_count": applied, - "tool_call_results": tool_calls, - }) - - self._log_turn_details( - player_action=player_action or last_prompt or "", - last_prompt=last_prompt or "", - strategy_name=strategy_name, - die_roll=die_roll, + on_thought=on_thought, + on_action=on_action, + on_player_roll=on_player_roll, + on_debug=on_debug, model=self.model, temperature=self.temperature, + timeout=self.timeout, max_tokens=self.max_tokens, - book_log=book_log, - log_entry=log_entry or "", - ambience=ambience, - tool_calls=tool_calls, - on_debug=on_debug, + api_key=self.api_key, + api_base=self.api_base, ) - return TurnResult( - book_log=book_log, - log_entry=log_entry, - user_prompt=user_prompt, - ambience=ambience, - debug_info="; ".join(phase3_errors) if phase3_errors else "", - changes=changes, - ) - - # ── Helpers ───────────────────────────────────────────────────────── - - @staticmethod - def _auto_prompt(book_log: str) -> str: - """Fallback player prompt.""" - return "**What do you do?**" - - def _validate_narrative(self, book_log: str, *, on_debug: callable = None) -> tuple[bool, str]: - """Check if book_log is acceptable narrative. Returns (ok, reason).""" - lines = book_log.strip().split("\n") - if not lines: - return False, "Empty narrative" - - common = Counter(lines).most_common(1) - if common and common[0][1] >= 5: - return False, f"Repetition: '{common[0][0][:60]}' ×{common[0][1]}" - - mech_lines = [l for l in lines if re.match( - r'^\*\*(?:Roll|Damage|Success|Failure|Check|Save|Hit|Miss|' - r'Strenght|Dexterity|Willpower|STR|DEX|WIL|' - r'(?:[A-Z][a-z]+(?: \(\w+\))?:))', - l - )] - if mech_lines: - ratio = len(mech_lines) / len(lines) - if ratio > 0.3: - return False, f"Game mechanics dominate ({len(mech_lines)}/{len(lines)} lines)" - - if re.search(r'```(?:tool|json)', book_log): - return False, "Contains unprocessed tool blocks" - - prose = re.sub(r'[*_#>`~\-\d]', '', book_log).strip() - if len(prose) < 50: - return False, "Too short to be meaningful" - - text = call_llm([ - {"role": "user", "content": - f"Rate this RPG narrative quality 1-5.\n" - f"1 = unreadable (spam, repetition, pure mechanics, garbled)\n" - f"2 = poor (mostly mechanics, little story)\n" - f"3 = acceptable (some narrative but rough)\n" - f"4 = good (solid prose, minor issues)\n" - f"5 = excellent (vivid, engaging)\n" - f"Reply with ONLY a single digit 1-5.\n\n" - f"{book_log[:600]}"} - ], model=self.model, temperature=self.temperature, timeout=self.timeout, - max_tokens=2, label="Narrative validation", on_debug=on_debug) - - if text and text.strip().isdigit(): - score = int(text.strip()) - if score < 3: - return False, f"Quality score: {score}/5" - - return True, "" - - # ── 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. - """ - if text.startswith('{"error":'): - try: - err = json.loads(text).get("error", "Unknown error") - except json.JSONDecodeError: - err = "Unknown error" - return GenerationResult(narrative="", error=err) - - 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_lines = [] - for line in narrative.splitlines(): - if not line.lstrip().startswith('book_log:'): - narrative_lines.append(line) - narrative = "\n".join(narrative_lines).strip() - try: - data = json.loads(json_str) - except json.JSONDecodeError: - pass - else: - 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", []), - ) - - # ── Logging ───────────────────────────────────────────────────────── - - def _log_turn_details( - self, - player_action: str, - last_prompt: str, - strategy_name: str, - die_roll: int, - model: str, - temperature: float, - max_tokens: int, - book_log: str, - log_entry: str, - ambience: Optional[str], - tool_calls: list, - on_debug, - ) -> None: - """Write structured turn summary to llm.log and fire TUI debug event.""" - ts = datetime.now().isoformat() - output_chars = len(book_log) - output_words = len(book_log.split()) if book_log else 0 - applied = len([tc for tc in tool_calls if tc.get("tool") != "finalize_turn"]) - - state.append_llm_log("") - state.append_llm_log(f"┌─ Turn Details — {ts}") - state.append_llm_log(f"├─ Input: {player_action}") - state.append_llm_log(f"├─ Last Prompt: {last_prompt}") - state.append_llm_log(f"├─ Strategy: {strategy_name}") - state.append_llm_log(f"├─ Dice: {die_roll} (1d6)") - state.append_llm_log(f"├─ Model: {model} | Temp: {temperature} | Tokens: {max_tokens}") - state.append_llm_log(f"├─ Output: {output_chars} chars ({output_words} words)") - state.append_llm_log(f"├─ Log Entry: {log_entry}") - state.append_llm_log(f"├─ Ambience: {ambience or 'None'}") - tools_preview = ", ".join(tc.get("tool", "?") for tc in tool_calls) - state.append_llm_log(f"├─ Tool Calls: {len(tool_calls)} ({tools_preview})") - state.append_llm_log( - "└─────────────────────────────────────────────────────────────────────────────────────────┘" - ) - - if on_debug: - on_debug("turn_details", { - "timestamp": ts, - "model": model, - "temperature": temperature, - "max_tokens": max_tokens, - "strategy_name": strategy_name, - "die_roll": die_roll, - "player_action": player_action, - "book_log_chars": output_chars, - "book_log_words": output_words, - "ambience": ambience, - "tool_calls_count": len(tool_calls), - "applied_changes_count": applied, - "tool_call_results": tool_calls, - }) # ── CLI entry point (for testing) ───────────────────────────────────────── diff --git a/tools/engine_lib/__init__.py b/tools/engine_lib/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/engine_lib/config.py b/tools/engine_lib/config.py new file mode 100644 index 0000000..5e887c0 --- /dev/null +++ b/tools/engine_lib/config.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +""" +config.py — LLM configuration loading and accessors for The Chaos engine. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +from .paths import CONFIG_PATH + +DEFAULT_CONFIG: dict = { + "llm": { + "model": "ollama/llama3.1", + "api_key": None, + "api_base": None, + "temperature": 0.8, + "max_tokens": 300, + } +} + + +def load_config(path: Path = CONFIG_PATH) -> dict: + """Load config from path, creating default if missing. Returns config dict.""" + if not 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, + ) + cfg = dict(DEFAULT_CONFIG) + save_config(cfg, path) + return cfg + raw = path.read_text() + cfg = json.loads(raw) + llm = cfg.get("llm", {}) + if not llm.get("api_key"): + llm["api_key"] = None + return cfg + + +def save_config(config: dict, path: Path = CONFIG_PATH) -> None: + """Save config dict to path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(config, indent=2) + "\n") + + +# ── Accessors ────────────────────────────────────────────────────────────── + +def get_model(config: dict) -> str: + return config.get("llm", {}).get("model", "ollama/llama3.1") + + +def get_api_key(config: dict) -> str | None: + return config.get("llm", {}).get("api_key") + + +def get_api_base(config: dict) -> str | None: + return config.get("llm", {}).get("api_base") + + +def get_temperature(config: dict) -> float: + return config.get("llm", {}).get("temperature", 0.8) + + +def get_max_tokens(config: dict) -> int: + return config.get("llm", {}).get("max_tokens", 512) + + +def get_timeout(config: dict) -> int: + return config.get("llm", {}).get("timeout", 120) diff --git a/tools/engine_lib/context.py b/tools/engine_lib/context.py new file mode 100644 index 0000000..668b5d9 --- /dev/null +++ b/tools/engine_lib/context.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +""" +context.py — System prompt and user message assembly for The Chaos engine. + +All functions are standalone — no dependency on GameEngine. +""" + +from __future__ import annotations + +from .paths import CHAR_PATH, WORLD_PATH, BOOK_PATH +from .prompts import SYSTEM_PROMPT +from . import state + + +def build_system_prompt() -> str: + """Assemble the system prompt with current game state.""" + char = state.read_file(CHAR_PATH) or "*No character sheet.*" + world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*" + log = state.read_recent_log() + story = state.read_recent_book() + return SYSTEM_PROMPT.substitute( + character=char, world=world, log=log, story=story + ) + + +def build_prose_prompt() -> str: + """Assemble the prose-generation prompt with current game state.""" + from .prompts import PROSE_PROMPT + char = state.read_file(CHAR_PATH) or "*No character sheet.*" + world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world state.*" + log = state.read_recent_log() + story = state.read_recent_book() + return PROSE_PROMPT.substitute( + character=char, world=world, log=log, story=story + ) + + +def build_user_message( + 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's Request\n{player_action}") + + has_existing_story = bool( + state.read_file(BOOK_PATH).strip() + ) if not last_prompt else True + + if not player_action and not last_prompt: + if has_existing_story: + raise RuntimeError("User action is required for every turn.") + else: + parts.append( + "## Instructions\n" + "This is a new story. Welcome the player and guide them through the game setup." + ) + else: + parts.append( + "## Instructions\n" + "Advance the story based on the player's request. " + "All state is shown above — write the outcome directly." + ) + return "\n\n".join(parts) diff --git a/tools/llm.py b/tools/engine_lib/llm.py similarity index 98% rename from tools/llm.py rename to tools/engine_lib/llm.py index d0ee38f..c05cecf 100644 --- a/tools/llm.py +++ b/tools/engine_lib/llm.py @@ -10,7 +10,7 @@ from __future__ import annotations import os -from state import append_llm_log +from .state import append_llm_log def set_llm_env(model: str, api_key: str | None, api_base: str | None) -> None: diff --git a/tools/models.py b/tools/engine_lib/models.py similarity index 100% rename from tools/models.py rename to tools/engine_lib/models.py diff --git a/tools/engine_lib/parsing.py b/tools/engine_lib/parsing.py new file mode 100644 index 0000000..57b00e4 --- /dev/null +++ b/tools/engine_lib/parsing.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +""" +parsing.py — LLM response parsing and turn logging for The Chaos engine. + +Standalone functions — no dependency on GameEngine. +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime +from typing import Optional + +from .models import GenerationResult +from . import state + + +def parse_response(text: str) -> GenerationResult: + """ + Parse a full LLM response into a GenerationResult. + Extracts the JSON block and splits narrative from it. + """ + if text.startswith('{"error":'): + try: + err = json.loads(text).get("error", "Unknown error") + except json.JSONDecodeError: + err = "Unknown error" + return GenerationResult(narrative="", error=err) + + 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_lines = [] + for line in narrative.splitlines(): + if not line.lstrip().startswith('book_log:'): + narrative_lines.append(line) + narrative = "\n".join(narrative_lines).strip() + try: + data = json.loads(json_str) + except json.JSONDecodeError: + pass + else: + 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", []), + ) + + +def log_turn_details( + player_action: str, + last_prompt: str, + strategy_name: str, + die_roll: int, + model: str, + temperature: float, + max_tokens: int, + book_log: str, + log_entry: str, + ambience: Optional[str], + tool_calls: list, + on_debug=None, +) -> None: + """Write structured turn summary to llm.log and fire TUI debug event.""" + ts = datetime.now().isoformat() + output_chars = len(book_log) + output_words = len(book_log.split()) if book_log else 0 + applied = len([tc for tc in tool_calls if tc.get("tool") != "finalize_turn"]) + + state.append_llm_log("") + state.append_llm_log(f"┌─ Turn Details — {ts}") + state.append_llm_log(f"├─ Input: {player_action}") + state.append_llm_log(f"├─ Last Prompt: {last_prompt}") + state.append_llm_log(f"├─ Strategy: {strategy_name}") + state.append_llm_log(f"├─ Dice: {die_roll} (1d6)") + state.append_llm_log(f"├─ Model: {model} | Temp: {temperature} | Tokens: {max_tokens}") + state.append_llm_log(f"├─ Output: {output_chars} chars ({output_words} words)") + state.append_llm_log(f"├─ Log Entry: {log_entry}") + state.append_llm_log(f"├─ Ambience: {ambience or 'None'}") + tools_preview = ", ".join(tc.get("tool", "?") for tc in tool_calls) + state.append_llm_log(f"├─ Tool Calls: {len(tool_calls)} ({tools_preview})") + state.append_llm_log( + "└─────────────────────────────────────────────────────────────────────────────────────────┘" + ) + + if on_debug: + on_debug("turn_details", { + "timestamp": ts, + "model": model, + "temperature": temperature, + "max_tokens": max_tokens, + "strategy_name": strategy_name, + "die_roll": die_roll, + "player_action": player_action, + "book_log_chars": output_chars, + "book_log_words": output_words, + "ambience": ambience, + "tool_calls_count": len(tool_calls), + "applied_changes_count": applied, + "tool_call_results": tool_calls, + }) diff --git a/tools/config_paths.py b/tools/engine_lib/paths.py similarity index 85% rename from tools/config_paths.py rename to tools/engine_lib/paths.py index 334eb2c..6f611a8 100644 --- a/tools/config_paths.py +++ b/tools/engine_lib/paths.py @@ -1,8 +1,6 @@ #!/usr/bin/env python3 """ -config_paths.py — Path constants for The Chaos game engine. - -Shared by engine.py, run.py, and all sub-modules. +paths.py — Path constants for The Chaos game engine. """ from __future__ import annotations diff --git a/tools/prompts.py b/tools/engine_lib/prompts.py similarity index 100% rename from tools/prompts.py rename to tools/engine_lib/prompts.py diff --git a/tools/state.py b/tools/engine_lib/state.py similarity index 99% rename from tools/state.py rename to tools/engine_lib/state.py index 85ecf43..f7d5584 100644 --- a/tools/state.py +++ b/tools/engine_lib/state.py @@ -13,12 +13,12 @@ import sys from datetime import date, datetime, timedelta from pathlib import Path -from paths import ( +from .paths import ( CHAR_PATH, WORLD_PATH, BOOK_PATH, JOURNAL_PATH, AMBIENCE_PATH, LOG_DIR, LLM_LOG_PATH, AMBIENCE_OPTIONS_PATH, CHANGES_PATH, AUDIO_DIR, TODAY, ) -from models import TurnResult +from .models import TurnResult def read_file(path: Path) -> str: diff --git a/tools/engine_lib/strategies.py b/tools/engine_lib/strategies.py new file mode 100644 index 0000000..0462f1a --- /dev/null +++ b/tools/engine_lib/strategies.py @@ -0,0 +1,657 @@ +#!/usr/bin/env python3 +""" +strategies.py — Generation strategies for The Chaos engine. + +Contains the three-phase conversational approach and the single-call +tool-based approach. All functions are standalone — no dependency on +GameEngine (config values and callbacks are passed explicitly). +""" + +from __future__ import annotations + +import json +import random +import re +from datetime import datetime +from typing import Iterator + +from .models import GenerationResult, TurnResult +from .prompts import PROSE_PROMPT +from .llm import set_llm_env, call_llm +from .tools_handler import ( + execute_tool, describe_tool_action, describe_change, + parse_changes_block, extract_tool_calls, +) +from .context import build_system_prompt, build_user_message, build_prose_prompt +from .validation import auto_prompt, validate_narrative +from .parsing import parse_response, log_turn_details +from . import state + + +# ── Synchronous (legacy) ─────────────────────────────────────────────────── + +def generate( + player_action: str | None = None, + last_narrative: str | None = None, + *, + model: str, + temperature: float, + timeout: int, + max_tokens: int, + api_key: str | None = None, + api_base: str | None = None, +) -> GenerationResult: + """ + Synchronous generation. Calls the LLM, parses the response, + and returns a GenerationResult. + """ + system = build_system_prompt() + user = 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_llm_env(model, api_key, api_base) + + try: + response = litellm.completion( + model=model, + messages=messages, + temperature=temperature, + stream=False, + timeout=timeout, + ) + text = response.choices[0].message.content or "" + except Exception as e: + return GenerationResult( + narrative="", + error=f"LLM call failed: {e}", + ) + + return parse_response(text) + + +# ── Streaming (legacy) ───────────────────────────────────────────────────── + +def generate_stream( + player_action: str | None = None, + last_narrative: str | None = None, + *, + model: str, + temperature: float, + timeout: int, + max_tokens: int, + api_key: str | None = None, + api_base: 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 = build_system_prompt() + user = 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 + + set_llm_env(model, api_key, api_base) + + try: + response = litellm.completion( + model=model, + messages=messages, + temperature=temperature, + stream=True, + timeout=timeout, + ) + full_text = "" + for chunk in response: + delta = chunk.choices[0].delta.content or "" + if delta: + full_text += delta + yield full_text + yield full_text + except Exception as e: + yield json.dumps({"error": f"LLM call failed: {e}"}) + + +# ── Three-phase generation ───────────────────────────────────────────────── + +def generate_with_tools( + 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, + *, + model: str, + temperature: float, + timeout: int, + max_tokens: int, + api_key: str | None = None, + api_base: str | None = None, +) -> TurnResult: + """ + Three-phase generation: + + 1. **Prose** — LLM writes the full book_log from context + player action. + 2. **Summarize** — LLM condenses the book_log into one log line. + 3. **Extract** — LLM reads the book_log and outputs tool calls for state changes. + """ + set_llm_env(model, api_key, api_base) + datetime_now = datetime.now() + state.append_llm_log(f"\n{'='*60}") + state.append_llm_log(f"=== Turn — {datetime_now.strftime('%Y-%m-%d %H:%M:%S')} ===") + state.append_llm_log(f"{'='*60}") + if player_action: + state.append_llm_log(f"Player: {player_action}") + elif last_prompt: + state.append_llm_log(f"Resume from: {last_prompt[:120]}") + + die_roll = random.randint(1, 6) + state.append_llm_log(f"Dice: {die_roll} (1d6)") + + book_log = None + changes_block = "" + log_entry = None + user_prompt = auto_prompt("") + ambience = None + debug_info = "" + changes: list[str] = [] + + for outer_attempt in range(3): + # ── Phase 1: Prose ──────────────────────────────────────────────── + if on_action: + on_action(f"Phase 1/3: writing story (dice={die_roll})") + if on_debug: + on_debug("phase", {"phase": 1, "name": "prose", "status": "start", "dice": die_roll, "outer_attempt": outer_attempt + 1}) + + system = build_prose_prompt() + + user = build_user_message( + player_action=player_action, + last_prompt=last_prompt, + ) + user += f"\n\n*A die is cast: **{die_roll}** (1d6).*" + + text = call_llm([ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], model=model, temperature=temperature, timeout=timeout, + max_tokens=1024, label=f"Prose attempt {outer_attempt + 1}", on_debug=on_debug) + + if not text or not text.strip(): + if on_debug: + on_debug("phase", {"phase": 1, "status": "empty", "attempt": outer_attempt + 1}) + continue + + raw = text.strip() + changes_block = "" + if "### Changes" in raw: + parts = raw.split("### Changes", 1) + book_log = parts[0].strip() + changes_block = "### Changes" + parts[1] + else: + book_log = raw + if on_debug: + preview = book_log[:150].replace("\n", "\\n") + on_debug("phase", {"phase": 1, "status": "done", "chars": len(book_log), "changes": bool(changes_block), "preview": preview}) + + # ── Validation ──────────────────────────────────────────────────── + if on_debug: + on_debug("phase", {"phase": 1, "name": "validation", "status": "start"}) + valid, reason = validate_narrative(book_log, model=model, temperature=temperature, timeout=timeout, on_debug=on_debug) + if not valid: + if on_debug: + on_debug("phase", {"phase": 1, "status": "validation_failed", "reason": reason, "outer_attempt": outer_attempt + 1}) + book_log = None + continue + + # ── Phase 2: Summarize ──────────────────────────────────────────── + if on_action: + on_action("Phase 2/3: summarizing story") + if on_debug: + on_debug("phase", {"phase": 2, "name": "summarize", "status": "start"}) + + log_context = state.read_recent_log() + log_entry = None + for p2_attempt in range(2): + context = book_log + if changes_block: + context += f"\n\n{changes_block}" + text = call_llm([ + {"role": "user", "content": + f"Given the session log so far, summarize the new story in one line. " + f"Focus on who was involved (character and NPC names):\n\n" + f"## Session Log\n{log_context}\n\n" + f"## New Story\n{context}"} + ], model=model, temperature=temperature, timeout=timeout, + max_tokens=max_tokens, label=f"Summarize attempt {p2_attempt + 1}", on_debug=on_debug) + if text and text.strip(): + log_entry = text.strip().split("\n")[0][:300] + if on_debug: + on_debug("phase", {"phase": 2, "status": "done", "summary": log_entry}) + break + + if not log_entry: + log_entry = book_log.split("\n")[0][:120] + if on_debug: + on_debug("phase", {"phase": 2, "status": "fallback", "summary": log_entry}) + + # ── Phase 3: Extract state changes ──────────────────────────────── + if on_action: + on_action("Phase 3/3: extracting state changes") + if on_debug: + on_debug("phase", {"phase": 3, "name": "extract", "status": "start"}) + + user_prompt = auto_prompt(book_log) + ambience = None + phase3_errors = [] + changes = [] + + # Step 1: Parse ### Changes block directly + if changes_block.strip(): + for tc in parse_changes_block(changes_block): + name = tc["tool"] + args = tc.get("args", {}) + if name == "finalize_turn": + continue + result = execute_tool(name, args) + if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"): + phase3_errors.append(f"{name}: {result}") + else: + desc = describe_change(name, args) + if desc: + changes.append(desc) + + # Step 2: LLM Phase 3 for finalize_turn + any extra changes + previous_attempt = None + phase3_ok = False + for p3_attempt in range(5): + from paths import CHAR_PATH, WORLD_PATH + current_char = state.read_file(CHAR_PATH) or "*No character.*" + current_world = state.truncate_world(state.read_file(WORLD_PATH) or "") or "*No world.*" + + phase3_prompt = ( + f"## Current Character\n{current_char}\n\n" + f"## Current World\n{current_world}\n\n" + f"## Story\n{book_log}\n\n" + ) + if changes_block.strip(): + phase3_prompt += ( + f"## Changes already applied\n{changes_block}\n\n" + f"Output the finalize_turn tool to end the turn. " + f"Add extra tool calls if you spot changes the list above missed.\n\n" + ) + else: + phase3_prompt += ( + f"Read the story and compare with current state. Output tool calls for any changes:\n\n" + ) + phase3_prompt += ( + f"Output ```tool blocks for changes only. Examples:\n\n" + ) + + if previous_attempt: + phase3_prompt += ( + f"--- PREVIOUS ATTEMPT (had errors) ---\n" + f"{previous_attempt['output']}\n\n" + f"--- FEEDBACK ---\n" + f"{previous_attempt['feedback']}\n\n" + f"Fix the issues above. Output corrected tool calls only.\n\n" + ) + + text = call_llm([ + {"role": "user", "content": phase3_prompt + + f"```tool\n{{\"tool\": \"modify_vitals\", \"args\": {{\"current_hp\": 5, \"cash\": 45}}}}\n```\n" + f"```tool\n{{\"tool\": \"modify_traits\", \"args\": {{\"dex\": 15}}}}\n```\n" + f"```tool\n{{\"tool\": \"add_to_inventory\", \"args\": {{\"item\": \"Silver key\"}}}}\n```\n" + f"```tool\n{{\"tool\": \"remove_from_inventory\", \"args\": {{\"item\": \"Torches (10)\"}}}}\n```\n" + f"```tool\n{{\"tool\": \"replace_gear\", \"args\": {{\"before\": \"Mace (1d6+1)\", \"after\": \"Mace (1d6+2, sharpened)\"}}}}\n```\n" + f"```tool\n{{\"tool\": \"add_note\", \"args\": {{\"note\": \"Found a hidden passage under the temple\"}}}}\n```\n" + f"```tool\n{{\"tool\": \"replace_note\", \"args\": {{\"before\": \"Old note text\", \"after\": \"New note text\"}}}}\n```\n" + f"```tool\n{{\"tool\": \"world_update\", \"args\": {{\"content\": \"# The World\\n\\n...full new world state...\"}}}}\n```\n" + f"```tool\n{{\"tool\": \"journal_update\", \"args\": {{\"add\": [\"Investigate the mine\"], \"done\": [\"Defeat the demon\"]}}}}\n```\n" + f"```tool\n{{\"tool\": \"finalize_turn\", \"args\": {{\"user_prompt\": \"What do you do?\", \"ambience\": \"dungeon\"}}}}\n```\n\n" + f"Only output tools for things that actually changed. Omit unchanged fields."} + ], model=model, temperature=temperature, timeout=timeout, + max_tokens=max_tokens, label=f"Extract attempt {p3_attempt + 1}", on_debug=on_debug) + + if not text or not text.strip(): + if on_debug: + on_debug("phase", {"phase": 3, "status": "empty", "attempt": p3_attempt + 1}) + continue + + tool_calls_list = extract_tool_calls( + text, round_num=p3_attempt + 1, on_debug=on_debug + ) + if on_debug and tool_calls_list: + names = [tc.get("tool", "?") for tc in tool_calls_list if tc.get("tool") != "finalize_turn"] + fin = any(tc.get("tool") == "finalize_turn" for tc in tool_calls_list) + on_debug("phase", {"phase": 3, "status": "tools_found", "tools": names, "has_finalize": fin}) + + errors: list[str] = [] + attempt_changes: list[str] = [] + for tc in tool_calls_list: + name = tc.get("tool", "?") + args = tc.get("args", {}) + if name == "finalize_turn": + if args.get("user_prompt"): + user_prompt = args["user_prompt"] + if args.get("ambience"): + ambience = args["ambience"] + continue + if on_action: + on_action(f"State: {describe_tool_action(name, args)}") + if on_debug: + on_debug("tool_call", {"round": p3_attempt + 1, "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 = execute_tool(name, args) + + if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"): + errors.append(f"{name}: {result}") + else: + desc = describe_change(name, args) + if desc: + attempt_changes.append(desc) + if on_debug: + on_debug("tool_result", {"round": p3_attempt + 1, "tool": name, "result": result}) + + if not errors: + phase3_ok = True + debug_info = "" + changes.extend(attempt_changes) + if on_debug: + on_debug("phase", {"phase": 3, "status": "done", "applied": len([tc for tc in tool_calls_list if tc.get("tool") != "finalize_turn"])}) + break + + phase3_errors = errors + debug_info = "; ".join(errors) + if on_debug: + on_debug("phase", {"phase": 3, "status": "errors", "errors": errors, "attempt": p3_attempt + 1}) + + feedback_lines = ["The previous tool calls had errors:"] + for e in errors: + feedback_lines.append(f"- {e}") + feedback_lines.append("") + feedback_lines.append("Fix ALL issues above. Use correct tool names, valid JSON, and reasonable values.") + previous_attempt = {"output": text, "feedback": "\n".join(feedback_lines)} + + if phase3_ok: + break + + if on_debug: + on_debug("phase", {"phase": 3, "status": "exhausted", "errors": phase3_errors}) + on_debug("phase", {"phase": 1, "status": "retry_after_phase3_failure", "outer_attempt": outer_attempt + 1}) + book_log = None + + if not book_log: + return TurnResult(error="Generation failed after exhausting all retries") + + if on_action: + on_action("Turn complete") + if on_debug: + on_debug("phase_done", { + "book_log_chars": len(book_log), + "log_entry": log_entry, + "user_prompt": user_prompt, + "ambience": ambience, + "extract_errors": debug_info or None, + }) + + state.append_llm_log( + f"\n--- FINAL ---\n" + f"book_log: {book_log[:200]}\n" + f"log_entry: {log_entry}\n" + f"user_prompt: {user_prompt}\n" + f"ambience: {ambience}\n" + ) + return TurnResult( + book_log=book_log, + log_entry=log_entry, + user_prompt=user_prompt, + ambience=ambience, + debug_info=debug_info, + changes=changes, + ) + + +# ── Single-call generation ───────────────────────────────────────────────── + +def generate_with_tools_single( + 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, + *, + model: str, + temperature: float, + timeout: int, + max_tokens: int, + api_key: str | None = None, + api_base: str | None = None, +) -> TurnResult: + """ + Single-call generation using tools. + + Uses a single LLM call with all tools available — LLM outputs + narrative + tool blocks in one go. No retry loop. + """ + datetime_now = datetime.now() + state.append_llm_log(f"\n{'='*60}") + state.append_llm_log(f"=== Turn — {datetime_now.strftime('%Y-%m-%d %H:%M:%S')} ===") + state.append_llm_log(f"{'='*60}") + if player_action: + state.append_llm_log(f"Player: {player_action}") + elif last_prompt: + state.append_llm_log(f"Resume from: {last_prompt[:120]}") + + strategy_name = "tools" + if on_action: + on_action(f"LLM: {model} | temp={temperature} | tokens={max_tokens} | strategy={strategy_name}") + if on_debug: + on_debug("config", {"model": model, "temperature": temperature, "max_tokens": max_tokens, "strategy": strategy_name}) + + die_roll = random.randint(1, 6) + state.append_llm_log(f"Dice: {die_roll} (1d6)") + + system = """You are an RPG dungeon master. The player just took an action. + +Output ONLY ```tool blocks — no prose, no reasoning, no explanation outside tool blocks. Every piece of output must be in a tool block. + +Use these tools to perform every action. Wrap each in its own ```tool block: +```tool +{"tool": "narrative", "args": {"text": "The full vivid narrative prose goes here."}} +``` +```tool +{"tool": "modify_vitals", "args": {"current_hp": 5, "cash": 45}} +``` +```tool +{"tool": "modify_traits", "args": {"dex": 15}} +``` +```tool +{"tool": "add_to_inventory", "args": {"item": "Silver key"}} +``` +```tool +{"tool": "remove_from_inventory", "args": {"item": "Torches (10)"}} +``` +```tool +{"tool": "replace_gear", "args": {"before": "Mace (1d6+1)", "after": "Mace (1d6+2, sharpened)"}} +``` +```tool +{"tool": "add_note", "args": {"note": "Found a hidden passage under the temple"}} +``` +```tool +{"tool": "replace_note", "args": {"before": "Old note text", "after": "New note text"}} +``` +```tool +{"tool": "world_update", "args": {"content": "# The World\n\n...full new world state..."}} +``` +```tool +{"tool": "journal_update", "args": {"add": ["Investigate the mine"], "done": ["Defeat the demon"]}} +``` +```tool +{"tool": "finalize_turn", "args": {"user_prompt": "What do you do?", "ambience": "dungeon"}} +``` +""" + system += build_prose_prompt() + + user = build_user_message( + player_action=player_action, + last_prompt=last_prompt, + ) + user += f"\n\n*A die is cast: **{die_roll}** (1d6).*" + + start_time = datetime.now() + set_llm_env(model, api_key, api_base) + state.append_llm_log(f"\n[TOOL] Single call — {len(system)} chars system, {len(user)} chars user") + state.append_llm_log(f"System preview: {system.split(chr(10))[0][:80]}...") + state.append_llm_log(f"User preview: {user.split(chr(10))[0][:80]}...") + + text = call_llm( + [{"role": "system", "content": system}, + {"role": "user", "content": user}], + model=model, temperature=temperature, timeout=timeout, + max_tokens=4096, label="Single tool call", on_debug=on_debug, + ) + + total_elapsed = (datetime.now() - start_time).total_seconds() * 1000 + if text: + state.append_llm_log(f"\n[TOOL] got {len(text)} chars in {total_elapsed:.1f}ms") + + if not text or not text.strip(): + return TurnResult(error="Single tool call returned empty response") + + raw = text.strip() + book_log = "" + log_entry = None + user_prompt = auto_prompt("") + ambience = None + tool_calls = [] + changes: list[str] = [] + phase3_errors: list[str] = [] + + tool_pattern = r"```tool\s*\n?(.*?)\n?```" + matches = re.findall(tool_pattern, text, re.DOTALL) + if matches: + for block in matches: + block = block.strip() + try: + tc = json.loads(block) + tool_calls.append(tc) + name = tc.get("tool", "unknown") + args = tc.get("args", {}) + state.append_llm_log(f"\n[EXTRACT] {name}: {json.dumps(args)[:100]}") + + if name == "narrative": + book_log = args.get("text", book_log) + elif name == "finalize_turn": + if args.get("user_prompt"): + user_prompt = args["user_prompt"] + if args.get("ambience"): + ambience = args["ambience"] + except json.JSONDecodeError as e: + state.append_llm_log(f"\n[EXTRACT] bad JSON: {e}") + continue + + log_entry = None + if book_log: + clean = re.sub(r'\s+', ' ', book_log).strip() + first_sentence = re.split(r'(?<=[.!?])\s+', clean) + if first_sentence: + log_entry = first_sentence[0].strip()[:200] + else: + log_entry = clean[:200] + state.append_llm_log(f"\n[SUMMARY] \"{log_entry}\"") + + extr_start = datetime.now() + for tc in tool_calls: + name = tc.get("tool", "unknown") + args = tc.get("args", {}) + if name in ("finalize_turn", "narrative"): + continue + result = execute_tool(name, args) + if result.startswith("**Error:") or result.startswith("Tool error") or result.startswith("Unknown"): + phase3_errors.append(f"{name}: {result}") + else: + desc = describe_change(name, args) + if desc: + changes.append(desc) + + apply_elapsed = (datetime.now() - extr_start).total_seconds() * 1000 + state.append_llm_log(f"\n[APPLY] {len(changes)} changes in {apply_elapsed:.1f}ms") + else: + state.append_llm_log(f"\n[TOOL] no tool blocks found") + + elapsed = (datetime.now() - start_time).total_seconds() * 1000 + + if on_action: + on_action("Turn complete") + if on_debug: + applied = len([tc for tc in tool_calls if tc.get("tool") != "finalize_turn"]) + on_debug("phase_done", { + "book_log_chars": len(book_log), + "log_entry": log_entry, + "user_prompt": user_prompt, + "ambience": ambience, + "extract_errors": phase3_errors or None, + "total_elapsed_ms": elapsed, + "tool_calls_count": len(tool_calls), + "applied_changes_count": applied, + "tool_call_results": tool_calls, + }) + + log_turn_details( + player_action=player_action or last_prompt or "", + last_prompt=last_prompt or "", + strategy_name=strategy_name, + die_roll=die_roll, + model=model, + temperature=temperature, + max_tokens=max_tokens, + book_log=book_log, + log_entry=log_entry or "", + ambience=ambience, + tool_calls=tool_calls, + on_debug=on_debug, + ) + return TurnResult( + book_log=book_log, + log_entry=log_entry, + user_prompt=user_prompt, + ambience=ambience, + debug_info="; ".join(phase3_errors) if phase3_errors else "", + changes=changes, + ) diff --git a/tools/tools_handler.py b/tools/engine_lib/tools_handler.py similarity index 99% rename from tools/tools_handler.py rename to tools/engine_lib/tools_handler.py index 80d7383..4fa8283 100644 --- a/tools/tools_handler.py +++ b/tools/engine_lib/tools_handler.py @@ -12,8 +12,8 @@ import json import random import re -from paths import CHAR_PATH, WORLD_PATH, LOG_DIR, TODAY -from state import read_file, validate_update_size, update_journal, append_llm_log +from .paths import CHAR_PATH, WORLD_PATH, LOG_DIR, TODAY +from .state import read_file, validate_update_size, update_journal, append_llm_log # ── Tool Registry ─────────────────────────────────────────────────────────── diff --git a/tools/engine_lib/validation.py b/tools/engine_lib/validation.py new file mode 100644 index 0000000..605f2c2 --- /dev/null +++ b/tools/engine_lib/validation.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +""" +validation.py — Narrative quality validation for The Chaos engine. + +Standalone functions — no dependency on GameEngine. +""" + +from __future__ import annotations + +import re +from collections import Counter + +from .llm import call_llm + + +def auto_prompt(book_log: str = "") -> str: + """Fallback player prompt.""" + return "**What do you do?**" + + +def validate_narrative( + book_log: str, + *, + model: str, + temperature: float, + timeout: int, + on_debug: callable = None, +) -> tuple[bool, str]: + """Check if book_log is acceptable narrative. Returns (ok, reason).""" + lines = book_log.strip().split("\n") + if not lines: + return False, "Empty narrative" + + common = Counter(lines).most_common(1) + if common and common[0][1] >= 5: + return False, f"Repetition: '{common[0][0][:60]}' ×{common[0][1]}" + + mech_lines = [l for l in lines if re.match( + r'^\*\*(?:Roll|Damage|Success|Failure|Check|Save|Hit|Miss|' + r'Strenght|Dexterity|Willpower|STR|DEX|WIL|' + r'(?:[A-Z][a-z]+(?: \(\w+\))?:))', + l + )] + if mech_lines: + ratio = len(mech_lines) / len(lines) + if ratio > 0.3: + return False, f"Game mechanics dominate ({len(mech_lines)}/{len(lines)} lines)" + + if re.search(r'```(?:tool|json)', book_log): + return False, "Contains unprocessed tool blocks" + + prose = re.sub(r'[*_#>`~\-\d]', '', book_log).strip() + if len(prose) < 50: + return False, "Too short to be meaningful" + + text = call_llm([ + {"role": "user", "content": + f"Rate this RPG narrative quality 1-5.\n" + f"1 = unreadable (spam, repetition, pure mechanics, garbled)\n" + f"2 = poor (mostly mechanics, little story)\n" + f"3 = acceptable (some narrative but rough)\n" + f"4 = good (solid prose, minor issues)\n" + f"5 = excellent (vivid, engaging)\n" + f"Reply with ONLY a single digit 1-5.\n\n" + f"{book_log[:600]}"} + ], model=model, temperature=temperature, timeout=timeout, + max_tokens=2, label="Narrative validation", on_debug=on_debug) + + if text and text.strip().isdigit(): + score = int(text.strip()) + if score < 3: + return False, f"Quality score: {score}/5" + + return True, "" diff --git a/tools/run.py b/tools/run.py index 93a846c..70df999 100755 --- a/tools/run.py +++ b/tools/run.py @@ -8,55 +8,35 @@ Owns the TUI and game loop. Layout: from __future__ import annotations import json -import os -import random -import sys import threading -from datetime import date -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 -# ── Game engine ───────────────────────────────────────── from engine import GameEngine -from models import GenerationResult, TurnResult -from paths import LLM_LOG_PATH -import state - -# ── Optional miniaudio ──────────────────────────────────── -try: - import miniaudio - HAS_AUDIO = True -except ImportError: - HAS_AUDIO = False - print("Note: miniaudio not installed — no ambience music.", file=sys.stderr) +from engine_lib.models import TurnResult +from engine_lib import state +from run_utils import ( + BOOK_PATH, CHAR_PATH, LAST_PROMPT_PATH, CHANGES_PATH, SETTINGS_PATH, + TODAY, REFRESH_SECS, clear_llm_log, ensure_log, + load_book_pages, +) +from run_ambience import AmbiencePlayer +from run_widgets import ( + app_ambience_player as _widget_player_ref, + RollModal, DebugPane, CharPane, StatusBar, TodoPane, TranscriptPane, +) -# ── Paths ──────────────────────────────────────────────── -BASE = Path(__file__).resolve().parent.parent -SESSION = BASE / 'session' -LOG_DIR = SESSION / 'log' -CHAR_PATH = SESSION / 'character.md' -WORLD_PATH = SESSION / 'world.md' -JOURNAL_PATH = SESSION / 'journal.md' -AMBIENCE_PATH = SESSION / 'ambience.md' -AMBIENCE_OPTIONS_PATH = SESSION / 'ambience_options.md' -BOOK_PATH = SESSION / 'book.md' -LAST_PROMPT_PATH = SESSION / 'last_prompt.md' -CHANGES_PATH = SESSION / 'changes.md' -SETTINGS_PATH = SESSION / 'settings.json' -AUDIO_DIR = SESSION / 'audio' -TODAY = date.today().isoformat() -LOG_PATH = LOG_DIR / f'{TODAY}.md' +# ── Global state ───────────────────────────────────────── +app_ambience_player: AmbiencePlayer | None = None -REFRESH_SECS = 2 +# ── UI Theme ───────────────────────────────────────────── MARKDOWN_THEME = Theme({ "markdown.h1": "bold #ff6b6b on #2a0000", "markdown.h2": "bold #ffd93d", @@ -66,532 +46,50 @@ MARKDOWN_THEME = Theme({ "markdown.code_block": "on #1e1e2e", "markdown.block_quote": "dim italic #8395a7", "markdown.link": "underline #48dbfb", - "markdown.item": "#ff9f43", - "markdown.em": "italic #ff9ff3", + "markdown.item": "#ff9ff3", "markdown.strong": "bold #feca57", "markdown.horizontal_rule": "dim #555555", }) -# ── Helpers (file reading, status, book, ambience) ─────── -def ensure_log(): - LOG_DIR.mkdir(parents=True, exist_ok=True) - if not LOG_PATH.exists(): - LOG_PATH.write_text(f"# Session Log — {TODAY}\n\n") - _populate_if_empty() - -def _populate_if_empty(): - content = LOG_PATH.read_text().strip() - if content and len(content.splitlines()) > 2: - return -def clear_llm_log(): - """Clear llm.log at start of app.""" - LLM_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) - LLM_LOG_PATH.write_text("") - - -def read_todo(): - """Read TODO items from journal.md.""" - if not JOURNAL_PATH.exists(): - return ["—— No journal yet ——"] - lines = LOG_PATH.read_text().splitlines() - return [l for l in lines if l.strip() and not l.startswith('#')][-n:] - - - -def read_log_tail(n=200): - """Read the tail of the session log.""" - if not LOG_PATH.exists(): - return [] - lines = LOG_PATH.read_text().splitlines() - return [l for l in lines if l.strip() and not l.startswith("#")][-n:] - - -def status_summary(): - if not CHAR_PATH.exists(): - return "no character" - lines = CHAR_PATH.read_text().splitlines() - name = "?" - health = "?" - for l in lines: - if l.startswith('**Name:**'): - name = l.split(':', 1)[1].strip().strip('_').strip('*') - if l.startswith('**Current Health:**'): - h = l.split(':', 1)[1].strip().strip('_').strip('*') - if h: - health = h - if l.startswith('**Max Health:**'): - m = l.split(':', 1)[1].strip().strip('_').strip('*') - if m and health == '?': - health = m - return f"{name} ❤ {health}" - -def log_count(): - return len(read_log_tail()) - -def load_book_pages(): - if not BOOK_PATH.exists() or not BOOK_PATH.read_text().strip(): - return ["*The story has not begun.*"] - text = BOOK_PATH.read_text().strip() - turns = text.split('\n## ') - pages = [] - for i, t in enumerate(turns): - pages.append(t if i == 0 else '## ' + t) - return pages or ["*The story has not begun.*"] - -def parse_ambience_options(): - if not AMBIENCE_OPTIONS_PATH.exists(): - return {} - options = {} - lines = AMBIENCE_OPTIONS_PATH.read_text().splitlines() - in_table = False - for line in lines: - s = line.strip() - if not s.startswith('|') or not s.endswith('|'): - in_table = False - continue - parts = [p.strip() for p in s.split('|')] - parts = [p for p in parts if p] - if len(parts) < 2: - continue - if not in_table: - in_table = True - continue - if all(c in '-:| ' for c in s): - continue - name = parts[0].lower() - files = [f.strip() for f in parts[1].split(',') if f.strip()] - paths = [AUDIO_DIR / f for f in files] - options[name] = paths - return options - - -# ── Ambience subsystem ─────────────────────────────────── -class AmbiencePlayer: - def __init__(self): - self.current_ambience = 'silence' - self._last_mtime = 0 - self._options = {} - self._device = None - self._stream = None - self._muted = False - self.load_options() - - @property - def available(self): - return HAS_AUDIO - - @property - def ambience_name(self): - return self.current_ambience - - @property - def is_muted(self): - return self._muted - - def toggle_mute(self): - self._muted = not self._muted - if self._muted: - self._stop() - else: - self._load_current() - - def load_options(self): - self._options = parse_ambience_options() - - def _stop(self): - if self._device: - try: - self._device.close() - except Exception: - pass - self._device = None - self._stream = None - - def poll(self): - if not HAS_AUDIO: - return - try: - mtime = os.path.getmtime(AMBIENCE_PATH) - except OSError: - return - if mtime == self._last_mtime: - return - self._last_mtime = mtime - try: - name = AMBIENCE_PATH.read_text().strip().lower() - except OSError: - return - # Save the name even when muted — will play on unmute - self.current_ambience = name - self._stop() - if not self._muted and name != 'silence' and name in self._options: - self._play_current() - - def _switch_to(self, name): - if name == self.current_ambience: - return - self.current_ambience = name - self._stop() - if self._muted or name == 'silence' or name not in self._options: - return - self._play_current() - - def _play_current(self): - tracks = self._options.get(self.current_ambience, []) - valid = [t for t in tracks if t.exists()] - if not valid: - return - track = random.choice(valid) - try: - self._stream = miniaudio.stream_file(str(track)) - self._device = miniaudio.PlaybackDevice() - self._device.start(self._stream) - except Exception: - self.current_ambience = None - - def _load_current(self): - """Called on unmute — replay current ambience if not silence.""" - if self.current_ambience and self.current_ambience != 'silence': - self._play_current() - - -# module-level ref -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): - raise NotImplementedError - - def on_mount(self): - clear_llm_log() - ensure_log() - self.load() - self.set_interval(REFRESH_SECS, self.load) - - -class TodoPane(AutoStatic): - def load(self): - items = read_todo() - self.update("\n".join(f" ☐ {i}" for i in items)) - - -class TranscriptPane(AutoStatic): - def load(self): - lines = read_log_tail() - display = "\n".join(lines[-80:]) - if lines: - display += "\n\n>>--- NOW --->" - self.update(display) - self.call_after_refresh(self._scroll_bottom) - - def _scroll_bottom(self): - if self.parent and hasattr(self.parent, 'scroll_end'): - self.parent.scroll_end(animate=False) - - -class DebugPane(Static): - """Scrolling log of LLM thoughts, tool calls, and results for this turn.""" - - MAX_LINES = 200 - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self._lines: list[str] = [] - - def append(self, text: str) -> None: - self._lines.append(text) - if len(self._lines) > self.MAX_LINES: - self._lines.pop(0) - self.update("\n".join(self._lines[-100:])) - self.call_after_refresh(self._scroll_bottom) - - def _scroll_bottom(self): - if self.parent and hasattr(self.parent, 'scroll_end'): - self.parent.scroll_end(animate=False) - - def clear(self) -> None: - self._lines.clear() - self.update("") - - -class CharPane(AutoStatic): - def load(self): - if not CHAR_PATH.exists(): - self.update("*No character sheet*") - return - self.update(RichMarkdown(CHAR_PATH.read_text().strip())) - - -class StatusBar(AutoStatic): - def load(self): - char = status_summary() - count = log_count() - todo = len(read_todo()) - music = "" - if not HAS_AUDIO: - music = " │ ♫ (install miniaudio)" - elif app_ambience_player: - name = app_ambience_player.ambience_name - music = f" │ ♫ {name}" - self.update(f"{char} │ {count} entries │ {todo} todo │ {TODAY}{music}") - - # ── The App ────────────────────────────────────────────── class ChaosTUI(App): TITLE = "The Chaos" CSS = """ - Screen { - background: #121212; - } - #banner { - dock: top; - height: 1; - background: #2a2a2a; - color: #e0ad4c; - text-align: center; - } - #main { - height: 100%; - background: #111111; - } - #todo-header { - background: #3a2a1a; - color: #e0b060; - text-style: bold; - padding: 0 1; - height: 1; - } - #todo-content { - background: #1a1510; - color: #d0b080; - padding: 0 1; - height: 5; - max-height: 5; - overflow-y: auto; - scrollbar-size-vertical: 2; - } - #main-tabs { - height: 1fr; - } - TabbedContent { - background: #1a1a2a; - } - VerticalScroll { - overflow-y: auto; - scrollbar-size-vertical: 2; - scrollbar-color: #555555; - scrollbar-color-hover: #777777; - scrollbar-color-active: #999999; - } - #char-content { - background: #1e1e2a; - color: #c0c0c0; - padding: 0 1; - } - #transcript { - background: #1a2a1a; - color: #c8c8c8; - padding: 0 1; - } - #debug-content { - background: #1a1a1a; - color: #88b0a0; - padding: 0 1; - } - #debug-content .dm-thought { - color: #c0a060; - } - #debug-content .dm-tool { - color: #60a0c0; - } - #debug-content .dm-result { - color: #80a080; - } - - /* Play tab */ - #play-narrative { - background: #161616; - color: #d8d8d8; - padding: 1 2; - height: auto; - } - #play-status { - background: #1a1a2a; - color: #e0b060; - padding: 0 2; - height: 1; - text-style: bold italic; - text-align: center; - } - #play-status.processing { - background: #2a1a0a; - color: #ffd93d; - } - #play-input { - height: 3; - background: #222222; - color: #e0d0c0; - border: solid #555555; - padding: 0 1; - } - #play-input:focus { - border: solid #e0ad4c; - } - #play-input:disabled { - background: #1a1a1a; - color: #666666; - border: solid #333333; - } - - /* Book tab */ - #book-header { - background: #2d2d2d; - color: #e0c080; - text-style: bold; - padding: 0 1; - height: 1; - } - #book-nav { - height: 3; - background: #222222; - align: center middle; - } - #book-nav Button { - width: 10; - margin: 0 1; - } - #book-nav Button:disabled { - color: #444444; - } - #book-nav Button:hover { - text-style: bold; - } - #book-nav-center { - height: 3; - width: 1fr; - } - #book-page-label { - height: 1; - color: #c0b090; - text-style: bold; - padding: 0 2; - text-align: center; - } - #book-progress { - height: 1; - background: #1a1a1a; - color: #e0b060; - padding: 0 2; - text-align: center; - } - #book-scroll { - height: 1fr; - } - #book-content { - background: #161616; - color: #d8d8d8; - padding: 0 2; - } - - #status-bar { - background: #222222; - color: #888888; - padding: 0 1; - height: 1; - text-style: italic; - } - #mute-btn { - dock: bottom; - width: 6; - height: 1; - background: #2a2a2a; - color: #888888; - border: none; - padding: 0 1; - min-width: 6; - margin: 0; - } - #mute-btn:hover { - background: #3a3a3a; - color: #cccccc; - } - #mute-btn.muted { - color: #ff6b6b; - text-style: bold; - } + Screen { background: #121212; } + #banner { dock: top; height: 1; background: #2a2a2a; color: #e0ad4c; text-align: center; } + #main { height: 100%; background: #111111; } + #todo-header { background: #3a2d23; color: #e0b060; padding: 0 1; height: 1; } + #todo-content { background: #1a1510; color: #d0b080; padding: 0 1; height: 5; max-height:5; overflow-y:auto; scrollbar-size-vertical:2; } + #main-tabs { height: 1fr; } + TabbedContent { background: #1a1a2a; } + VerticalScroll { overflow-y: auto; scrollbar-size-vertical:2; scrollbar-color:#555; scrollbar-color-hover:#777; scrollbar-color-active:#999; } + #char-content { background: #1e1e2a; color: #c0c0c0; padding: 0 1; } + #transcript { background: #1a2a1a; color: #c8c8c8; padding: 0 1; } + #debug-content { background: #1a1a1a; color: #88b0a0; padding: 0 1; } + #debug-content .dm-thought { color: #c0a060; } + #debug-content .dm-tool { color: #60a0c0; } + #debug-content .dm-result { color: #80a080; } + #play-narrative { background: #161616; color: #d8d8d8; padding: 1 2; height: auto; } + #play-status { background: #1a2a1a; color: #e0b060; padding: 0 2; height: 1; text-style: bold italic; text-align: center; } + #play-status.processing { background: #2a1a0a; color: #ffd93d; } + #play-input { height: 3; background: #222; color: #e0d0c0; border: solid #555; padding: 0 1; } + #play-input:focus { border: solid #e0ad4c; } + #play-input:disabled { background: #1a1a1a; color: #666; border: solid #333; } + #book-header { background: #2d2d2d; color: #e0c080; text-style: bold; padding: 0 1; height: 1; } + #book-nav { height: 3; background: #222; align: center middle; } + #book-nav Button { width: 10; margin: 0 1; } + #book-nav Button:disabled { color: #444; } + #book-nav Button:hover { text-style: bold; } + #book-nav-center { height: 3; width: 1fr; } + #book-page-label { height: 1; color: #c0b090; text-style: bold; padding: 0 2; text-align: center; } + #book-progress { height: 1; background: #1a1a1a; color: #e0b060; padding: 0 2; text-align: center; } + #book-scroll { height: 1fr; } + #book-content { background: #161616; color: #d8d8d8; padding: 0 2; } + #status-bar { background: #222; color: #888; padding: 0 1; height: 1; text-style: italic; } + #mute-btn { dock: bottom; width: 6; height: 1; background: #2a2a2a; color: #888; border: none; padding: 0 1; min-width: 6; margin: 0; } + #mute-btn:hover { background: #3a3a3a; color: #ccc; } + #mute-btn.muted { color: #ff6b6b; text-style: bold; } """ BINDINGS = [ @@ -602,41 +100,25 @@ class ChaosTUI(App): def __init__(self, *args, no_music=False, **kwargs): super().__init__(*args, **kwargs) global app_ambience_player - if not no_music: - app_ambience_player = AmbiencePlayer() - else: - app_ambience_player = None + app_ambience_player = None if no_music else AmbiencePlayer() + import run_widgets + run_widgets.app_ambience_player = app_ambience_player - # Game engine self.engine = GameEngine() - - # Game loop state self._last_prompt: str = "" self._last_result: TurnResult | None = None self._is_processing: bool = False - - # Thinking animation self._spinner_frames = ["◴", "◷", "◶", "◵"] self._thinking_frame = 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 - self._book_pages = [] + self._book_pages: list[str] = [] self._prev_page_count = 0 - - # Debug log - self._debug_lines: list[str] = [] - - # Settings guard — prevent save during init before restore self._settings_loaded = False - # ── Settings persistence ────────────────────────────── def _load_settings(self) -> dict: defaults = {"active_tab": "play-tab", "music_muted": False, "book_page": 0} if SETTINGS_PATH.exists(): @@ -651,8 +133,7 @@ class ChaosTUI(App): def _save_settings(self) -> None: if not self._settings_loaded: return - tabs = self.query_one("#main-tabs", TabbedContent) - active = tabs.active if tabs else "play-tab" + active = self.query_one("#main-tabs", TabbedContent).active or "play-tab" data = { "active_tab": active, "music_muted": app_ambience_player.is_muted if app_ambience_player else False, @@ -663,7 +144,6 @@ class ChaosTUI(App): except OSError: pass - # ── Compose ────────────────────────────────────────── def compose(self): yield Static(f"⚔ The Chaos ╎ {TODAY}", id="banner") with Vertical(id="main"): @@ -674,10 +154,7 @@ class ChaosTUI(App): with VerticalScroll(id="play-scroll"): yield Static("*Awaiting the fates...*", id="play-narrative") yield Static("", id="play-status") - yield Input( - placeholder="Type your action and press Enter...", - id="play-input", - ) + yield Input(placeholder="Type your action and press Enter...", id="play-input") with TabPane("CHARACTER", id="char-tab"): with VerticalScroll(): yield CharPane(id="char-content") @@ -710,11 +187,9 @@ class ChaosTUI(App): self.set_interval(REFRESH_SECS, self._reload_book) self.call_after_refresh(self._update_mute_button) self.call_after_refresh(self._apply_settings) - # Start the game self.call_after_refresh(self._begin_game) def _apply_settings(self) -> None: - """Load and apply persisted UI settings.""" settings = self._load_settings() self._book_page = settings.get("book_page", 0) self._prev_page_count = len(self._book_pages) @@ -723,14 +198,12 @@ class ChaosTUI(App): app_ambience_player.toggle_mute() self._update_mute_button() try: - tabs = self.query_one("#main-tabs", TabbedContent) - tabs.active = settings.get("active_tab", "play-tab") + self.query_one("#main-tabs", TabbedContent).active = settings.get("active_tab", "play-tab") except Exception: pass self._settings_loaded = True def _begin_game(self): - """Resume from last saved prompt or generate an opening scene.""" if LAST_PROMPT_PATH.exists(): saved = LAST_PROMPT_PATH.read_text().strip() if saved: @@ -740,7 +213,7 @@ class ChaosTUI(App): if pages: parts.append(pages[-1]) if CHANGES_PATH.exists(): - changes = [l for l in CHANGES_PATH.read_text().splitlines() if l.strip()] + changes = [l.strip() for l in CHANGES_PATH.read_text().splitlines() if l.strip()] if changes: changes_text = "\n".join(f"> {c}" for c in changes) parts.append(f"> **Last turn changes:**\n{changes_text}") @@ -750,7 +223,6 @@ class ChaosTUI(App): return self._call_llm() - # ── Ambience ───────────────────────────────────────── def _check_ambience(self): if app_ambience_player: app_ambience_player.poll() @@ -773,39 +245,22 @@ class ChaosTUI(App): btn.classes = "" btn.tooltip = "Mute music" - # ── Game Loop ───────────────────────────────────────── def _call_llm(self, player_action: str | None = None): - """Called when the player has acted — sends their action to the LLM.""" if self._is_processing: return self._is_processing = True - input_widget = self.query_one("#play-input", Input) input_widget.disabled = True input_widget.placeholder = "DM is at work..." - self._show_thinking() - - # Clear debug for new turn - pane = self.query_one("#debug-content", DebugPane) - pane.clear() - if player_action: - self._append_debug(f"▶ player action: {player_action}") - else: - self._append_debug("▶ starting new turn") - - # Run generation in a daemon thread so it doesn't block the UI - t = threading.Thread( - target=self._run_generation, - args=(player_action,), - daemon=True, - ) + self.query_one("#debug-content", DebugPane).clear() + self._append_debug(f"▶ {'player action: ' + player_action if player_action else 'starting new turn'}") + t = threading.Thread(target=self._run_generation, args=(player_action,), daemon=True) t.start() def _run_generation(self, player_action: str | None) -> None: - """Worker thread: calls engine.generate_with_tools() or generate_with_tools_single() based on configured strategy.""" import traceback - last_prompt = self._last_prompt if self._last_prompt else None + last_prompt = self._last_prompt or None def on_thought(thought: str) -> None: self.call_from_thread(self._on_thought, thought) @@ -818,52 +273,35 @@ class ChaosTUI(App): try: strategy = self.engine.config.get("llm", {}).get("strategy", "tools") - if strategy == "tools": - result = self.engine.generate_with_tools_single( - player_action=player_action, - last_prompt=last_prompt, - on_thought=on_thought, - on_action=on_action, - on_player_roll=self._on_player_roll, - on_debug=on_debug, - ) - else: - result = self.engine.generate_with_tools( - player_action=player_action, - last_prompt=last_prompt, - on_thought=on_thought, - on_action=on_action, - on_player_roll=self._on_player_roll, - on_debug=on_debug, - ) + gen = (self.engine.generate_with_tools_single if strategy == "tools" + else self.engine.generate_with_tools) + result = gen( + player_action=player_action, + last_prompt=last_prompt, + on_thought=on_thought, + on_action=on_action, + on_player_roll=self._on_player_roll, + on_debug=on_debug, + ) except Exception as e: - tb = traceback.format_exc() - self.call_from_thread(self._on_generation_error, e, tb) + self.call_from_thread(self._on_generation_error, e, traceback.format_exc()) return - self.call_from_thread(self._on_generation_done, result, player_action) def _append_debug(self, text: str) -> None: - """Append a line to the debug pane.""" from datetime import datetime ts = datetime.now().strftime("%H:%M:%S") - pane = self.query_one("#debug-content", DebugPane) - pane.append(f"[{ts}] {text}") + self.query_one("#debug-content", DebugPane).append(f"[{ts}] {text}") def _show_thinking(self) -> None: - """Show the thinking indicator and start the animation timer.""" self._dm_action = "DM is weaving the narrative" self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") - spinner = self._spinner_frames[0] - status.update(f"✦ {spinner} {self._dm_action} ✦") - self._thinking_timer_handle = self.set_interval( - 0.5, self._tick_thinking - ) + status.update(f"✦ {self._spinner_frames[0]} {self._dm_action} ✦") + self._thinking_timer_handle = self.set_interval(0.5, self._tick_thinking) def _hide_thinking(self) -> None: - """Stop the animation and hide the thinking indicator.""" if self._thinking_timer_handle: self._thinking_timer_handle.stop() self._thinking_timer_handle = None @@ -872,71 +310,60 @@ class ChaosTUI(App): 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_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") - spinner = self._spinner_frames[0] - status.update(f"✦ {spinner} {display} ✦") + status.update(f"✦ {self._spinner_frames[0]} {display} ✦") self._append_debug(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_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") - spinner = self._spinner_frames[0] - status.update(f"✦ {spinner} {action} ✦") + status.update(f"✦ {self._spinner_frames[0]} {action} ✦") self._append_debug(action) def _on_debug(self, event_type: str, data: dict) -> None: - """Structured debug entry: visible description + technical detail.""" if event_type == "phase": p = data.get("phase", 0) status = data.get("status", "") if status == "start": - name = data.get("name", "") - dice = data.get("dice") - outer = data.get("outer_attempt") - d = f" dice={dice}" if dice else "" - o = f" [attempt {outer}/3]" if outer else "" - self._append_debug(f"▸ Phase {p}: {name}{o} {d}") + n = data.get("name", "") + d = f" dice={data['dice']}" if data.get("dice") else "" + o = f" [attempt {data['outer_attempt']}/3]" if data.get("outer_attempt") else "" + self._append_debug(f"▸ Phase {p}: {n}{o}{d}") elif status == "done": if p == 1: self._append_debug(f" ✔ prose: {data.get('chars', 0)} chars") elif p == 2: self._append_debug(f" ✔ summary: {data.get('summary', '')}") elif p == 3: - n = data.get("applied", 0) - self._append_debug(f" ✔ extract: {n} state changes applied") + self._append_debug(f" ✔ extract: {data.get('applied', 0)} state changes applied") elif status == "empty": self._append_debug(f" ⚠ phase {p} attempt {data.get('attempt', '?')} empty — retry") elif status == "fallback": self._append_debug(f" ⚠ phase {p} used fallback: {data.get('summary', '')}") elif status == "tools_found": tools = data.get("tools", []) - fin = data.get("has_finalize", False) t = ", ".join(tools) if tools else "none" - self._append_debug(f" 🔧 tools found: {t}" + (" + finalize_turn" if fin else "")) + self._append_debug(f" 🔧 tools: {t}" + (" + finalize" if data.get("has_finalize") else "")) elif status == "errors": - errs = data.get("errors", []) - for e in errs: + for e in data.get("errors", []): self._append_debug(f" ✖ {e}") - self._append_debug(f" ⟳ retry (attempt {data.get('attempt', '?')})") + self._append_debug(f" ⟳ retry ({data.get('attempt', '?')})") elif status == "exhausted": - errs = data.get("errors", []) - self._append_debug(f" ✖ Phase 3 exhausted all retries — state changes may be missing!") - for e in errs: + self._append_debug(" ✖ Phase 3 exhausted retries — state changes may be missing!") + for e in data.get("errors", []): self._append_debug(f" {e}") elif status == "retry_after_phase3_failure": - self._append_debug(f" ⟳ Phase 3 failed — retrying from Phase 1 (attempt {data.get('outer_attempt', '?')}/3)") + self._append_debug(f" ⟳ Phase 3 failed — retry from Phase 1 (attempt {data.get('outer_attempt', '?')}/3)") elif status == "validation_failed": self._append_debug(f" ✖ narrative rejected: {data.get('reason', '?')} (attempt {data.get('outer_attempt', '?')}/3)") elif event_type == "phase_done": - self._append_debug(f" ✔ turn complete — book_log: {data.get('book_log_chars', 0)} chars") + self._append_debug(f" ✔ turn — book_log: {data.get('book_log_chars', 0)} chars") if data.get("log_entry"): self._append_debug(f" log: {data['log_entry']}") if data.get("ambience"): @@ -944,55 +371,44 @@ class ChaosTUI(App): if data.get("extract_errors"): self._append_debug(f" extract errors: {data['extract_errors']}") elif event_type == "config": - model = data.get("model", "?") - temp = data.get("temperature", "?") - tokens = data.get("max_tokens", "?") - strat = data.get("strategy", "?") - self._append_debug(f"▸ LLM: {model} | temp={temp} | max_tokens={tokens} | strategy={strat}") + m = data.get("model", "?") + t = data.get("temperature", "?") + tk = data.get("max_tokens", "?") + s = data.get("strategy", "?") + self._append_debug(f"▸ LLM: {m} | temp={t} | tokens={tk} | strategy={s}") elif event_type == "tool_call": - tool = data.get("tool", "?") - args = data.get("args", {}) - self._append_debug(f" 🔧 {tool}({json.dumps(args)})") + self._append_debug(f" 🔧 {data['tool']}({json.dumps(data.get('args', {}))})") elif event_type == "tool_result": - result = data.get("result", "") - preview = result[:80].replace("\n", " ").strip() + ("…" if len(result) > 80 else "") - self._append_debug(f" → {preview}") + r = data.get("result", "") + self._append_debug(f" → {r[:80].replace(chr(10),' ').strip()}{'…' if len(r)>80 else ''}") elif event_type == "parse_error": self._append_debug(f" ⚠ bad tool block: {data.get('content', '')}") elif event_type == "llm_error": - label = data.get("label", "") - err = data.get("error", "") - self._append_debug(f" ✖ LLM error [{label}]: {err}") + self._append_debug(f" ✖ LLM [{data.get('label','')}]: {data.get('error','')}") elif event_type == "turn_details": ts = data.get("timestamp", "") - model = data.get("model", "?") - temp = data.get("temperature", "?") - tokens = data.get("max_tokens", "?") - strat = data.get("strategy_name", "?") - dice = data.get("die_roll", "?") - input_chars = len(data.get("player_action", "")) - output_chars = data.get("book_log_chars", 0) - words = data.get("book_log_words", 0) - ambience = data.get("ambience", "None") - tool_count = data.get("tool_calls_count", 0) - applied = data.get("applied_changes_count", 0) - total_ms = data.get("total_elapsed_ms", 0) - apply_ms = data.get("apply_elapsed_ms", 0) + m = data.get("model", "?") + t = data.get("temperature", "?") + tk = data.get("max_tokens", "?") + s = data.get("strategy_name", "?") + d = data.get("die_roll", "?") + ic = len(data.get("player_action", "")) + oc = data.get("book_log_chars", 0) + w = data.get("book_log_words", 0) + a = data.get("ambience", "None") + tc = data.get("tool_calls_count", 0) + ap = data.get("applied_changes_count", 0) + tms = data.get("total_elapsed_ms", 0) + ams = data.get("apply_elapsed_ms", 0) self._append_debug(f" ━━━ Turn ━━━ {ts}") - self._append_debug(f" LLM: {model} | temp={temp} | tokens={tokens} | strategy={strat}") - self._append_debug(f" Dice: {dice} | Input: {input_chars} chars | Output: {output_chars} chars ({words} words)") - self._append_debug(f" Ambience: {ambience}") - self._append_debug(f" Tools: {tool_count} (applied: {applied})") - self._append_debug(f" Time: {total_ms:.0f}ms total — {apply_ms:.0f}ms apply") - self._append_debug(f" ━━━ Details ━━━") - for tc in data.get("tool_call_results", []): - tool = tc.get("tool", "?") - args = tc.get("args", {}) - self._append_debug(f" 🔧 {tool}: {json.dumps(args)[:120]}") + self._append_debug(f" LLM: {m} | temp={t} | tokens={tk} | strategy={s}") + self._append_debug(f" Dice: {d} | In: {ic}c | Out: {oc}c ({w}w)") + self._append_debug(f" Amb: {a} | Tools: {tc} (applied: {ap}) | Time: {tms:.0f}ms ({ams:.0f}ms apply)") + for tc2 in data.get("tool_call_results", []): + self._append_debug(f" 🔧 {tc2['tool']}: {json.dumps(tc2.get('args',{}))[:120]}") 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._append_debug, f"🎲 asks player to roll {dice} ({reason})") + self.call_from_thread(self._append_debug, f"🎲 roll {dice} ({reason})") self.call_from_thread(self._show_roll_modal, dice, reason) self._roll_event.wait() self._roll_event.clear() @@ -1001,7 +417,6 @@ class ChaosTUI(App): 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 @@ -1012,134 +427,79 @@ class ChaosTUI(App): self.push_screen(RollModal(dice, reason), on_dismiss) def _tick_thinking(self) -> None: - """Animate the spinner on the current DM action.""" if not self._is_processing: return 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"✦ {spinner} {self._dm_action} ✦") + self.query_one("#play-status", Static).update( + f"✦ {self._spinner_frames[self._thinking_frame]} {self._dm_action} ✦") - def _on_generation_done( - self, result: TurnResult, player_action: str | None - ) -> None: - """Handle the completed turn on the main thread.""" + def _on_generation_done(self, result: TurnResult, player_action: str | None) -> None: self._is_processing = False self._hide_thinking() - if result.error: self._show_error(result.error, result.debug_info) self._append_debug(f"✖ error: {result.error}") return - - # Log only after successful finalize — failed turns produce no side effects from datetime import datetime ts = datetime.now().strftime("%H:%M") if result.log_entry: state.append_log(f"- **{ts}** — {result.log_entry}") elif result.book_log: - first_line = result.book_log.strip().split("\n")[0][:80] - state.append_log(f"- **Turn** — {first_line}") - - # Archive the turn's book log + state.append_log(f"- **Turn** — {result.book_log.strip().split(chr(10))[0][:80]}") if result.book_log: state.archive_turn(result.book_log) - - # Apply state changes state.apply_state(result) - - # Display the next user prompt self._display_scene(result) - - # Persist the prompt so the game resumes here on restart if result.user_prompt: LAST_PROMPT_PATH.write_text(result.user_prompt.strip()) - - # Store for next turn self._last_prompt = result.user_prompt self._last_result = result self._append_debug("✔ turn complete") - def _on_generation_error( - self, error: Exception, traceback_str: str - ) -> None: - """Handle an unhandled exception from the worker thread.""" - import traceback + def _on_generation_error(self, error: Exception, traceback_str: str) -> None: self._is_processing = False self._hide_thinking() err_msg = f"{type(error).__name__}: {error}" - self._append_debug(f"✖ UNHANDLED EXCEPTION: {err_msg}") + self._append_debug(f"✖ UNHANDLED: {err_msg}") for line in traceback_str.rstrip().split("\n")[-10:]: self._append_debug(f" {line}") self._show_error(err_msg, traceback_str) def _display_scene(self, result: TurnResult) -> None: - """Update the UI with the last story entry followed by the DM prompt.""" parts = [] if result.book_log: parts.append(result.book_log) if result.changes: - changes_text = "\n".join(f"> {c}" for c in result.changes) - parts.append(f"> **Changes:**\n{changes_text}") + parts.append(f"> **Changes:**\n" + "\n".join(f"> {c}" for c in result.changes)) if result.user_prompt: parts.append(f"---\n\n{result.user_prompt}") self._set_narrative("\n\n".join(parts) if parts else "") self._enable_input() def _enable_input(self) -> None: - input_widget = self.query_one("#play-input", Input) - input_widget.disabled = False - input_widget.placeholder = "Type your action and press Enter..." - input_widget.value = "" - input_widget.focus() + inp = self.query_one("#play-input", Input) + inp.disabled = False + inp.placeholder = "Type your action and press Enter..." + inp.value = "" + inp.focus() def _set_narrative(self, text: str) -> None: - widget = self.query_one("#play-narrative", Static) - widget.update(RichMarkdown(text)) - # Scroll to top - scroll = self.query_one("#play-scroll", VerticalScroll) - scroll.scroll_home(animate=False) + self.query_one("#play-narrative", Static).update(RichMarkdown(text)) + self.query_one("#play-scroll", VerticalScroll).scroll_home(animate=False) 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) + t = f"**Error:** {error}\n\n" + (f"**Debug Info:**\n\n{debug_info}\n\n" if debug_info else "") + self._set_narrative(t + "Check your session/config.json and ensure your LLM provider is running.") self._enable_input() - # ── Input handling ──────────────────────────────────── def on_input_submitted(self, event: Input.Submitted) -> None: - """Player pressed Enter in the input widget.""" action = event.value.strip() if not action or self._is_processing: event.stop() return event.stop() - self._handle_player_action(action) - - def _handle_player_action(self, action: str) -> None: - """Handle a player action typed in the input.""" self._call_llm(player_action=action) - def _guess_time_of_day(self) -> str: - """Simple time-of-day label based on hour.""" - from datetime import datetime - h = datetime.now().hour - if h < 6: - return "Night" - elif h < 12: - return "Morning" - elif h < 14: - return "Midday" - elif h < 18: - return "Afternoon" - elif h < 21: - return "Evening" - else: - return "Night" - - # ── Book viewer ─────────────────────────────────────── def _init_book(self): self._reload_book() self._render_book_page() @@ -1155,17 +515,11 @@ class ChaosTUI(App): def _render_book_page(self): if not self._book_pages: return - self.query_one("#book-content").update( - RichMarkdown(self._book_pages[self._book_page]) - ) + self.query_one("#book-content").update(RichMarkdown(self._book_pages[self._book_page])) total = len(self._book_pages) - self.query_one("#book-page-label").update( - f"Page {self._book_page + 1} of {total}" - ) + self.query_one("#book-page-label").update(f"Page {self._book_page + 1} of {total}") pct = (self._book_page + 1) / total if total else 1 - fill = round(pct * 20) - bar = "█" * fill + "░" * (20 - fill) - self.query_one("#book-progress").update(f"[{bar}]") + self.query_one("#book-progress").update(f"[{'█'*round(pct*20)}{'░'*(20-round(pct*20))}]") self.query_one("#book-prev").disabled = (self._book_page == 0) self.query_one("#book-next").disabled = (self._book_page == total - 1) @@ -1195,31 +549,11 @@ class ChaosTUI(App): self._save_settings() - -def read_todo(): - """Read TODO items from journal.md.""" - if not JOURNAL_PATH.exists(): - return ["—— No journal yet ——"] - lines = JOURNAL_PATH.read_text().splitlines() - in_todo = False - todo = [] - for l in lines: - if l.strip().lstrip("#").strip().startswith("TODO"): - in_todo = True - continue - if l.strip().startswith("#") and in_todo: - break - if in_todo and l.strip(): - todo.append(l.strip().lstrip("- ")) - return todo or ["—— All done! ——"] - def main(): import argparse parser = argparse.ArgumentParser(description="The Chaos TUI") parser.add_argument('--no-music', action='store_true', help='Disable ambience music') - args = parser.parse_args() - app = ChaosTUI(no_music=args.no_music) - app.run() + ChaosTUI(no_music=parser.parse_args().no_music).run() if __name__ == '__main__': diff --git a/tools/run_ambience.py b/tools/run_ambience.py new file mode 100644 index 0000000..ca293ce --- /dev/null +++ b/tools/run_ambience.py @@ -0,0 +1,97 @@ +import os +import random +from run_utils import AMBIENCE_PATH, AMBIENCE_OPTIONS_PATH, AUDIO_DIR, parse_ambience_options + +try: + import miniaudio + HAS_AUDIO = True +except ImportError: + HAS_AUDIO = False + + +class AmbiencePlayer: + def __init__(self): + self.current_ambience = 'silence' + self._last_mtime = 0 + self._options = {} + self._device = None + self._stream = None + self._muted = False + self.load_options() + + @property + def available(self): + return HAS_AUDIO + + @property + def ambience_name(self): + return self.current_ambience + + @property + def is_muted(self): + return self._muted + + def toggle_mute(self): + self._muted = not self._muted + if self._muted: + self._stop() + else: + self._load_current() + + def load_options(self): + self._options = parse_ambience_options() + + def _stop(self): + if self._device: + try: + self._device.close() + except Exception: + pass + self._device = None + self._stream = None + + def poll(self): + if not HAS_AUDIO: + return + try: + mtime = os.path.getmtime(AMBIENCE_PATH) + except OSError: + return + if mtime == self._last_mtime: + return + self._last_mtime = mtime + try: + name = AMBIENCE_PATH.read_text().strip().lower() + except OSError: + return + self.current_ambience = name + self._stop() + if not self._muted and name != 'silence' and name in self._options: + self._play_current() + + def _switch_to(self, name): + if name == self.current_ambience: + return + self.current_ambience = name + self._stop() + if self._muted or name == 'silence' or name not in self._options: + return + self._play_current() + + def _play_current(self): + tracks = self._options.get(self.current_ambience, []) + valid = [t for t in tracks if t.exists()] + if not valid: + return + track = random.choice(valid) + try: + self._stream = miniaudio.stream_file(str(track)) + self._device = miniaudio.PlaybackDevice() + self._device.start(self._stream) + except Exception: + self.current_ambience = None + self._stop() + + def _load_current(self): + if self.current_ambience and self.current_ambience != 'silence': + self._play_current() diff --git a/tools/run_utils.py b/tools/run_utils.py new file mode 100644 index 0000000..30abe5c --- /dev/null +++ b/tools/run_utils.py @@ -0,0 +1,125 @@ +from pathlib import Path +from datetime import date + +BASE = Path(__file__).resolve().parent.parent +SESSION = BASE / 'session' +LOG_DIR = SESSION / 'log' +CHAR_PATH = SESSION / 'character.md' +WORLD_PATH = SESSION / 'world.md' +JOURNAL_PATH = SESSION / 'journal.md' +AMBIENCE_PATH = SESSION / 'ambience.md' +AMBIENCE_OPTIONS_PATH = SESSION / 'ambience_options.md' +BOOK_PATH = SESSION / 'book.md' +LAST_PROMPT_PATH = SESSION / 'last_prompt.md' +CHANGES_PATH = SESSION / 'changes.md' +SETTINGS_PATH = SESSION / 'settings.json' +AUDIO_DIR = SESSION / 'audio' +TODAY = date.today().isoformat() +LOG_PATH = LOG_DIR / f'{TODAY}.md' + +REFRESH_SECS = 2 + + +def ensure_log(): + LOG_DIR.mkdir(parents=True, exist_ok=True) + if not LOG_PATH.exists(): + LOG_PATH.write_text(f"# Session Log — {TODAY}\n\n") + _populate_if_empty() + + +def _populate_if_empty(): + content = LOG_PATH.read_text().strip() + if content and len(content.splitlines()) > 2: + return + + +def clear_llm_log(): + from engine_lib.paths import LLM_LOG_PATH + LLM_LOG_PATH.parent.mkdir(parents=True, exist_ok=True) + LLM_LOG_PATH.write_text("") + + +def read_todo(): + if not JOURNAL_PATH.exists(): + return ["—— No journal yet ——"] + lines = JOURNAL_PATH.read_text().splitlines() + in_todo = False + todo = [] + for l in lines: + if l.strip().lstrip("#").strip().startswith("TODO"): + in_todo = True + continue + if l.strip().startswith("#") and in_todo: + break + if in_todo and l.strip(): + todo.append(l.strip().lstrip("- ")) + return todo or ["—— All done! ——"] + + +def read_log_tail(n=200): + if not LOG_PATH.exists(): + return [] + lines = LOG_PATH.read_text().splitlines() + return [l for l in lines if l.strip() and not l.startswith("#")][-n:] + + +def status_summary(): + if not CHAR_PATH.exists(): + return "no character" + lines = CHAR_PATH.read_text().splitlines() + name = "?" + health = "?" + for l in lines: + if l.startswith('**Name:**'): + name = l.split(':', 1)[1].strip().strip('_').strip('*') + if l.startswith('**Current Health:**'): + h = l.split(':', 1)[1].strip().strip('_').strip('*') + if h: + health = h + if l.startswith('**Max Health:**'): + m = l.split(':', 1)[1].strip().strip('_').strip('*') + if m and health == '?': + health = m + return f"{name} ❤ {health}" + + +def log_count(): + return len(read_log_tail()) + + +def load_book_pages(): + if not BOOK_PATH.exists() or not BOOK_PATH.read_text().strip(): + return ["*The story has not begun.*"] + text = BOOK_PATH.read_text().strip() + turns = text.split('\n## ') + pages = [] + for i, t in enumerate(turns): + pages.append(t if i == 0 else '## ' + t) + return pages or ["*The story has not begun.*"] + + +def parse_ambience_options(): + if not AMBIENCE_OPTIONS_PATH.exists(): + return {} + options = {} + lines = AMBIENCE_OPTIONS_PATH.read_text().splitlines() + in_table = False + for line in lines: + s = line.strip() + if not s.startswith('|') or not s.endswith('|'): + in_table = False + continue + parts = [p.strip() for p in s.split('|')] + parts = [p for p in parts if p] + if len(parts) < 2: + continue + if not in_table: + in_table = True + continue + if all(c in '-:| ' for c in s): + continue + name = parts[0].lower() + files = [f.strip() for f in parts[1].split(',') if f.strip()] + paths = [AUDIO_DIR / f for f in files] + options[name] = paths + return options diff --git a/tools/run_widgets.py b/tools/run_widgets.py new file mode 100644 index 0000000..5859ca1 --- /dev/null +++ b/tools/run_widgets.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import Screen +from textual.widgets import Button, Input, Static +from rich.markdown import Markdown as RichMarkdown + +from run_utils import ( + CHAR_PATH, TODAY, REFRESH_SECS, + clear_llm_log, ensure_log, read_todo, read_log_tail, + status_summary, log_count, +) +from run_ambience import HAS_AUDIO + + +# module-level ref filled by ChaosTUI +app_ambience_player: object | None = None + + +class RollModal(Screen): + 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": + self._submit(self.query_one("#roll-input", Input).value) + + def _submit(self, value: str) -> None: + val = value.strip() + if val: + self.dismiss(val) + + +class AutoStatic(Static): + def load(self): + raise NotImplementedError + + def on_mount(self): + clear_llm_log() + ensure_log() + self.load() + self.set_interval(REFRESH_SECS, self.load) + + +class TodoPane(AutoStatic): + def load(self): + items = read_todo() + self.update("\n".join(f" ☐ {i}" for i in items)) + + +class TranscriptPane(AutoStatic): + def load(self): + lines = read_log_tail() + display = "\n".join(lines[-80:]) + if lines: + display += "\n\n>>--- NOW --->" + self.update(display) + self.call_after_refresh(self._scroll_bottom) + + def _scroll_bottom(self): + if self.parent and hasattr(self.parent, 'scroll_end'): + self.parent.scroll_end(animate=False) + + +class DebugPane(Static): + MAX_LINES = 200 + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._lines: list[str] = [] + + def append(self, text: str) -> None: + self._lines.append(text) + if len(self._lines) > self.MAX_LINES: + self._lines.pop(0) + self.update("\n".join(self._lines[-100:])) + self.call_after_refresh(self._scroll_bottom) + + def _scroll_bottom(self): + if self.parent and hasattr(self.parent, 'scroll_end'): + self.parent.scroll_end(animate=False) + + def clear(self) -> None: + self._lines.clear() + self.update("") + + +class CharPane(AutoStatic): + def load(self): + if not CHAR_PATH.exists(): + self.update("*No character sheet*") + return + self.update(RichMarkdown(CHAR_PATH.read_text().strip())) + + +class StatusBar(AutoStatic): + def load(self): + char = status_summary() + count = log_count() + todo = len(read_todo()) + music = "" + if not HAS_AUDIO: + music = " │ ♫ (install miniaudio)" + elif app_ambience_player: + name = app_ambience_player.ambience_name + music = f" │ ♫ {name}" + self.update(f"{char} │ {count} entries │ {todo} todo │ {TODAY}{music}") diff --git a/tools/test_imports.py b/tools/test_imports.py index af01b6a..ac49ef2 100755 --- a/tools/test_imports.py +++ b/tools/test_imports.py @@ -7,12 +7,17 @@ import ast MODULES = [ 'engine.py', - 'config_paths.py', - 'models.py', - 'prompts.py', - 'state.py', - 'tools_handler.py', - 'llm.py', + 'engine_lib/paths.py', + 'engine_lib/models.py', + 'engine_lib/prompts.py', + 'engine_lib/config.py', + 'engine_lib/context.py', + 'engine_lib/state.py', + 'engine_lib/tools_handler.py', + 'engine_lib/llm.py', + 'engine_lib/validation.py', + 'engine_lib/parsing.py', + 'engine_lib/strategies.py', ] def check_missing_imports(): diff --git a/tools/test_runtime.py b/tools/test_runtime.py index f0e2f54..792a1b0 100755 --- a/tools/test_runtime.py +++ b/tools/test_runtime.py @@ -22,12 +22,17 @@ def test_engine_import(): sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) modules_to_test = [ - ('paths', ['BASE_DIR', 'SESSION_DIR', 'CHAR_PATH', 'LLM_LOG_PATH']), - ('models', ['GenerationResult', 'TurnResult']), - ('prompts', ['SYSTEM_PROMPT', 'PROSE_PROMPT']), - ('state', ['read_file', 'apply_state', 'append_log', 'append_llm_log']), - ('tools_handler', ['execute_tool', 'extract_tool_calls', 'TOOL_REGISTRY']), - ('llm', ['call_llm', 'set_llm_env']), + ('engine_lib.paths', ['BASE_DIR', 'SESSION_DIR', 'CHAR_PATH', 'LLM_LOG_PATH']), + ('engine_lib.models', ['GenerationResult', 'TurnResult']), + ('engine_lib.prompts', ['SYSTEM_PROMPT', 'PROSE_PROMPT']), + ('engine_lib.config', ['load_config', 'save_config', 'get_model']), + ('engine_lib.context', ['build_system_prompt', 'build_user_message', 'build_prose_prompt']), + ('engine_lib.state', ['read_file', 'apply_state', 'append_log', 'append_llm_log']), + ('engine_lib.tools_handler', ['execute_tool', 'extract_tool_calls', 'TOOL_REGISTRY']), + ('engine_lib.llm', ['call_llm', 'set_llm_env']), + ('engine_lib.validation', ['validate_narrative', 'auto_prompt']), + ('engine_lib.parsing', ['parse_response', 'log_turn_details']), + ('engine_lib.strategies', ['generate_with_tools', 'generate_with_tools_single']), ('engine', ['GameEngine']), ]