diff --git a/tools/engine.py b/tools/engine.py index f0ef1f6..93e0a83 100644 --- a/tools/engine.py +++ b/tools/engine.py @@ -1,84 +1,27 @@ -#!/usr/bin/env python3 -""" -engine.py — The Chaos Game Engine - -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 random +import re import sys +from datetime import datetime from pathlib import Path -from engine_lib.paths import CONFIG_PATH -from engine_lib.models import GenerationResult, TurnResult +from engine_lib.models import TurnResult from engine_lib import config -from engine_lib import strategies +from engine_lib.context import build_system_prompt +from engine_lib.validation import validate_action, auto_prompt +from engine_lib.tools_handler import execute_tool, describe_change, extract_tool_calls +from engine_lib.parsing import log_turn_details +from engine_lib import state +from engine_lib.llm import call_llm class GameEngine: - """Owns configuration and delegates generation to standalone strategies.""" - def __init__(self, session_dir: str | Path | None = None): - from engine_lib.paths import SESSION_DIR - self.session_dir = Path(session_dir) if session_dir else SESSION_DIR - self.config = config.load_config(CONFIG_PATH) + self.config = config.load_config() - # ── Config accessors ──────────────────────────────────────────────── - - @property - def model(self) -> str: - return config.get_model(self.config) - - @property - def api_key(self) -> str | None: - return config.get_api_key(self.config) - - @property - def api_base(self) -> str | None: - return config.get_api_base(self.config) - - @property - def temperature(self) -> float: - return config.get_temperature(self.config) - - @property - def max_tokens(self) -> int: - return config.get_max_tokens(self.config) - - @property - def timeout(self) -> int: - return config.get_timeout(self.config) - - # ── Generation (delegated) ────────────────────────────────────────── - - def generate( - self, - player_action: str | None = None, - last_narrative: str | None = None, - ) -> GenerationResult: - 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, - ) - - 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, - ) - - def generate_with_tools( + def generate_turn( self, player_action: str | None = None, last_prompt: str | None = None, @@ -87,43 +30,162 @@ class GameEngine: on_player_roll: callable = None, on_debug: callable = None, ) -> TurnResult: - 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, + now = datetime.now() + state.append_llm_log(f"\n{'='*60}") + state.append_llm_log(f"=== Turn — {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)") + + lm = self.config.get("llm", {}) + model = lm.get("model", "ollama/llama3.1") + + if on_action: + on_action(f"LLM: {model} | temp={lm.get('temperature')}") + if on_debug: + on_debug("config", {"model": model, "temperature": lm.get("temperature"), "max_tokens": lm.get("max_tokens"), "strategy": "tools"}) + + if player_action: + valid, reason = validate_action(player_action, on_debug=on_debug) + if not valid: + state.append_llm_log(f"\n[VALIDATION REJECTED] {reason}") + return TurnResult( + book_log=f"You can't do that — {reason}.", + log_entry=f"You can't do that — {reason}.", + user_prompt=auto_prompt(""), + ) + + system = build_system_prompt() + parts = [] + if last_prompt: + parts.append(f"## Situation\n{last_prompt}") + if player_action: + parts.append(f"## Player's Request\n{player_action}") + if not player_action and not last_prompt: + 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." + ) + parts.append(f"\n*A die is cast: **{die_roll}** (1d6).*") + user = "\n\n".join(parts) + + start_time = datetime.now() + state.append_llm_log(f"\n[TOOL] Single call — {len(system)} chars system, {len(user)} chars user") + + text = call_llm( + [{"role": "system", "content": system}, {"role": "user", "content": user}], + label="Turn generation", on_debug=on_debug, - model=self.model, - temperature=self.temperature, - timeout=self.timeout, - max_tokens=self.max_tokens, ) - def generate_with_tools_single( - self, - player_action: str | None = None, - last_prompt: str | None = None, - on_thought: callable = None, - on_action: callable = None, - on_player_roll: callable = None, - on_debug: callable = None, - ) -> TurnResult: - return strategies.generate_with_tools_single( - player_action=player_action, - last_prompt=last_prompt, - on_thought=on_thought, - on_action=on_action, - on_player_roll=on_player_roll, + if not text or not text.strip(): + return TurnResult(error="LLM returned empty response") + + raw = text.strip() + state.append_llm_log(f"\n[TOOL] got {len(raw)} chars in {(datetime.now() - start_time).total_seconds() * 1000:.1f}ms") + + tool_calls = extract_tool_calls(raw, on_debug=on_debug) + if not tool_calls: + state.append_llm_log("\n[TOOL] no tool blocks found") + + book_log = "" + log_entry = None + user_prompt = auto_prompt("") + ambience = None + changes: list[str] = [] + errors: list[str] = [] + + extr_start = datetime.now() + + for tc in tool_calls: + name = tc.get("tool", "") + args = tc.get("args", {}) + + 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"] + elif 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 name not in ("narrative", "finalize_turn", "player_roll"): + 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: + changes.append(desc) + + if book_log: + clean = re.sub(r'\s+', ' ', book_log).strip() + sentences = re.split(r'(?<=[.!?])\s+', clean) + log_entry = sentences[0][:200] if sentences else clean[:200] + + apply_elapsed = (datetime.now() - extr_start).total_seconds() * 1000 + state.append_llm_log(f"\n[APPLY] {len(changes)} changes in {apply_elapsed:.1f}ms") + + total_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") not in ("finalize_turn", "narrative")]) + on_debug("phase_done", { + "book_log_chars": len(book_log), + "log_entry": log_entry, + "user_prompt": user_prompt, + "ambience": ambience, + "extract_errors": errors or None, + "total_elapsed_ms": total_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="tools", + die_roll=die_roll, + model=model, + temperature=lm.get("temperature", 0.8), + max_tokens=lm.get("max_tokens", 4096), + book_log=book_log, + log_entry=log_entry or "", + ambience=ambience, + tool_calls=tool_calls, on_debug=on_debug, - model=self.model, - temperature=self.temperature, - timeout=self.timeout, - max_tokens=self.max_tokens, + ) + + return TurnResult( + book_log=book_log, + log_entry=log_entry, + user_prompt=user_prompt, + ambience=ambience, + debug_info="; ".join(errors) if errors else "", + changes=changes, ) -# ── CLI entry point (for testing) ───────────────────────────────────────── def main(): """Generate a turn from the command line (debug/testing).""" import argparse @@ -134,7 +196,7 @@ def main(): args = parser.parse_args() engine = GameEngine() - result = engine.generate_with_tools_single( + result = engine.generate_turn( player_action=args.action, last_prompt=args.last, ) diff --git a/tools/engine_lib/context.py b/tools/engine_lib/context.py index 668b5d9..44fdc7d 100644 --- a/tools/engine_lib/context.py +++ b/tools/engine_lib/context.py @@ -1,13 +1,6 @@ -#!/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 .paths import CHAR_PATH, WORLD_PATH from .prompts import SYSTEM_PROMPT from . import state @@ -21,54 +14,3 @@ def build_system_prompt() -> str: 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/engine_lib/models.py b/tools/engine_lib/models.py index 503ba65..f5e8cf0 100644 --- a/tools/engine_lib/models.py +++ b/tools/engine_lib/models.py @@ -1,31 +1,12 @@ -#!/usr/bin/env python3 -""" -models.py — Data classes for The Chaos game engine. -""" - from __future__ import annotations from dataclasses import dataclass, field from typing import Optional -@dataclass -class GenerationResult: - """Legacy result — kept for backward compat with CLI main().""" - narrative: str - choices: list[str] = field(default_factory=list) - log_entry: Optional[str] = None - ambience: Optional[str] = None - character_updates: Optional[str] = None - world_updates: Optional[str] = None - journal_add: list[str] = field(default_factory=list) - journal_done: list[str] = field(default_factory=list) - error: Optional[str] = None - - @dataclass class TurnResult: - """Output of a complete turn via finalize_turn tool.""" + """Output of a complete turn.""" book_log: str = "" user_prompt: str = "" ambience: Optional[str] = None diff --git a/tools/engine_lib/parsing.py b/tools/engine_lib/parsing.py index 57b00e4..6278f21 100644 --- a/tools/engine_lib/parsing.py +++ b/tools/engine_lib/parsing.py @@ -1,72 +1,11 @@ -#!/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, @@ -85,7 +24,7 @@ def log_turn_details( 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"]) + applied = len([tc for tc in tool_calls if tc.get("tool") not in ("finalize_turn", "narrative")]) state.append_llm_log("") state.append_llm_log(f"┌─ Turn Details — {ts}") diff --git a/tools/engine_lib/prompts.py b/tools/engine_lib/prompts.py index d946fee..e3efbb7 100644 --- a/tools/engine_lib/prompts.py +++ b/tools/engine_lib/prompts.py @@ -1,13 +1,6 @@ -#!/usr/bin/env python3 -""" -prompts.py — System prompt templates for The Chaos game engine. -""" - from __future__ import annotations - from string import Template - SYSTEM_PROMPT = Template("""You are the DM for "The Chaos". Narrate in 2nd person ("You"), vivid but concise. Player: Dillion. ## Rules @@ -17,21 +10,53 @@ SYSTEM_PROMPT = Template("""You are the DM for "The Chaos". Narrate in 2nd perso - **Wounds at 0 HP**: 1d6 → 1-2 die, 3-4 -1 max HP, 5-6 -1 all rolls until healed. - **Modifiers**: Favourable +1, Risky -1, Desperate -2. -## Tools (action only) -Wrap in ```tool to perform an action: +## Output Format +Output ONLY ```tool blocks — no prose, no reasoning, no explanation outside tool blocks. Every piece of output must be in a tool block. + +```tool +{"tool": "narrative", "args": {"text": "The full vivid narrative prose goes here."}} ``` + +Wrap each action in its own ```tool block: + +```tool {"tool": "roll", "args": {"dice": "1d6"}} ``` +```tool +{"tool": "player_roll", "args": {"dice": "1d6", "reason": "a check"}} +``` +```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"}} +``` -- **roll** — dice, modifier -- **player_roll** — dice, reason -- **character_update** — content: "full sheet" (if HP/cash/gear/stats change) -- **world_update** — content: "full world" (if NPCs/locations/threads change) -- **journal_update** — add: [...], done: [...] - -You have the full state above — no need to look anything up. Just write the story and use tools when the player's action changes something. - -You are the sole authority over the game state. The player's action is a **proposal**, not a fact. If their action contradicts the character sheet (e.g. using an item they don't have, spending cash they don't have, claiming stats/abilities they don't have, or asserting events that didn't happen), narrate the failure and DO NOT use any state-changing tools. The character sheet is the single source of truth. +You are the sole authority over the game state. The player's action is a **proposal**, not a fact. If their action contradicts the character sheet (e.g. using an item they don't have, spending cash they don't have, claiming stats they don't have), narrate the failure and do NOT call any state-changing tools. **Inventory rule**: If the player wants to use an item, you must first verify it's on their character sheet. If it is, you MUST call `remove_from_inventory` for that item AND apply the effects (e.g. `modify_vitals` for HP potions). If it's not on the sheet, reject the action — do not let them use items they don't have. @@ -48,47 +73,3 @@ $log ### Story $story""") - - -PROSE_PROMPT = Template("""You are the DM for "The Chaos". Narrate in 2nd person ("You"), vivid but concise. Player: Dillion. - -## Rules -- **Odds**: 1d6, 4+ favourable, 3- trouble. -- **Traits**: 3d6, roll UNDER trait. -- **Combat**: 1d6, 4+ hits. Damage: 1d6 + mod - armour. -- **Wounds at 0 HP**: 1d6 → 1-2 die, 3-4 -1 max HP, 5-6 -1 all rolls until healed. -- **Modifiers**: Favourable +1, Risky -1, Desperate -2. - -A die is cast at the start of each turn — incorporate it into your narrative. - -End your response with a `### Changes` block listing what changed: - -### Changes -- Current Health: 3 -- Cash: 45 silver -- Added to inventory: Silver key -- Removed from inventory: Torches (10) -- Replaced gear: Mace (1d6+1) → Mace (1d6+2) -- Note: Found a hidden passage -- Journal done: Defeat the demon -- Journal add: Investigate the mine - -You are the sole authority over the game state. The player's action is a **proposal**, not a fact. If their action contradicts the character sheet (e.g. using an item they don't have), do NOT include any change lines and instead narrate the failure. - -**Inventory rule**: If the player wants to use an item, verify it's on the character sheet. If it is, include `- Removed from inventory: ` and any other relevant change lines (e.g. `- Current Health: `). If it's not on the sheet, reject the action — no change lines. - -Only include lines for things that actually changed. Omit unused lines entirely. - -## State - -### Character -$character - -### World -$world - -### Log -$log - -### Story -$story""") diff --git a/tools/engine_lib/strategies.py b/tools/engine_lib/strategies.py deleted file mode 100644 index f983bf8..0000000 --- a/tools/engine_lib/strategies.py +++ /dev/null @@ -1,683 +0,0 @@ -#!/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 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, validate_action -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, -) -> 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", - ) - - 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, -) -> 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 - - 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, -) -> 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. - """ - 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)") - - # ── Pre-generation validation ──────────────────────────────────── - if player_action: - valid, reason = validate_action( - player_action, - model=model, - timeout=timeout, - on_debug=on_debug, - ) - if not valid: - state.append_llm_log(f"\n[VALIDATION REJECTED] {reason}") - fail_narrative = f"You can't do that — {reason}." - return TurnResult( - book_log=fail_narrative, - log_entry=fail_narrative, - user_prompt=auto_prompt(""), - ) - - 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, -) -> 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. - -You are the sole authority over the game state. The player's action is a **proposal**, not a fact. If their action contradicts the character sheet (e.g. using an item they don't have, spending cash they don't have, claiming stats they don't have), narrate the failure with the narrative tool and do NOT call any state-changing tools. - -**Inventory rule**: If the player wants to use an item, verify it's on the character sheet first. If it is, you MUST call `remove_from_inventory` for that item AND apply effects (e.g. `modify_vitals`). If it's not on the sheet, narrate the failure — do not let them use items they don't have. - -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).*" - - # ── Pre-generation validation ──────────────────────────────────── - if player_action: - valid, reason = validate_action( - player_action, - model=model, - timeout=timeout, - on_debug=on_debug, - ) - if not valid: - state.append_llm_log(f"\n[VALIDATION REJECTED] {reason}") - fail_narrative = f"You can't do that — {reason}." - return TurnResult( - book_log=fail_narrative, - log_entry=fail_narrative, - user_prompt=auto_prompt(""), - ) - - start_time = datetime.now() - 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/engine_lib/tools_handler.py b/tools/engine_lib/tools_handler.py index 4fa8283..def5aae 100644 --- a/tools/engine_lib/tools_handler.py +++ b/tools/engine_lib/tools_handler.py @@ -1,11 +1,3 @@ -#!/usr/bin/env python3 -""" -tools_handler.py — Tool call infrastructure for The Chaos engine. - -Handles tool call extraction, execution, and description. All functions -are standalone — no dependency on the GameEngine class. -""" - from __future__ import annotations import json @@ -16,8 +8,6 @@ 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 ─────────────────────────────────────────────────────────── - TOOL_REGISTRY: dict[str, dict] = { "roll": {"description": "Roll dice.", "args": {"dice": "1d6", "modifier": "+1"}}, "player_roll": {"description": "Ask player to roll.", "args": {"dice": "1d6", "reason": "why"}}, @@ -34,8 +24,6 @@ TOOL_REGISTRY: dict[str, dict] = { } -# ── Character Sheet Patcher ───────────────────────────────────────────────── - def patch_character(pattern: str, repl: str, count: int = 1, flags: int = 0) -> str: """Apply a regex replacement to character.md. Returns error msg or empty string.""" text = CHAR_PATH.read_text() @@ -46,27 +34,7 @@ def patch_character(pattern: str, repl: str, count: int = 1, flags: int = 0) -> return "" -# ── Individual Tool Implementations ───────────────────────────────────────── - -def tool_think(args: dict) -> str: - return "" - - -def tool_read_file(args: dict) -> str: - filename = (args or {}).get("file", "") - paths = { - "character": CHAR_PATH, - "world": WORLD_PATH, - "log": LOG_DIR / f"{TODAY}.md", - } - path = paths.get(filename) - if not path: - return f"Unknown file: {filename}. Choose from: {', '.join(paths)}" - return read_file(path) or f"*{filename} is empty.*" - - def tool_roll(args: dict) -> str: - import random dice_str = (args or {}).get("dice", "1d6") modifier_str = (args or {}).get("modifier", "0") try: @@ -200,8 +168,6 @@ def tool_journal_update(args: dict) -> str: return "Journal updated." -# ── Tool Dispatcher ───────────────────────────────────────────────────────── - def execute_tool(tool_name: str, args: dict) -> str: """Execute a tool by name. Returns result string.""" fn_map = { @@ -228,59 +194,6 @@ def execute_tool(tool_name: str, args: dict) -> str: return f"Tool error ({tool_name}): {e}" -# ── Descriptions ──────────────────────────────────────────────────────────── - -def describe_tool_action(tool_name: str, args: dict) -> str: - """Return a user-facing status message for a tool call.""" - dm_status = (args or {}).get("dm_status") - if dm_status: - return f"DM is {dm_status}..." - - read_descriptions = { - "character": "reading the character sheet", - "world": "consulting the world map", - "book": "reviewing the story so far", - "log": "checking the session log", - "journal": "scanning the journal", - } - if tool_name == "read_file": - file = (args or {}).get("file", "") - desc = read_descriptions.get(file, f"reading {file}") - elif tool_name in ("character_get", "world_get", "journal_get"): - file = tool_name.replace("_get", "") - desc = read_descriptions.get(file, f"reading {file}") - elif tool_name in ("character_update", "world_update"): - desc = "updating the records" - elif tool_name == "journal_update": - desc = "updating the journal" - elif tool_name == "roll": - dice = (args or {}).get("dice", "1d6") - mod = (args or {}).get("modifier") - desc = f"rolling {dice}" - if mod: - desc += f" {mod}" - elif tool_name == "player_roll": - dice = (args or {}).get("dice", "1d6") - desc = f"asking you to roll {dice}" - elif tool_name == "modify_traits": - desc = "updating traits" - elif tool_name == "modify_vitals": - desc = "updating vitals" - elif tool_name == "add_to_inventory": - desc = "adding item to inventory" - elif tool_name == "remove_from_inventory": - desc = "removing item from inventory" - elif tool_name == "replace_gear": - desc = "replacing gear" - elif tool_name == "add_note": - desc = "adding note" - elif tool_name == "replace_note": - desc = "replacing note" - else: - desc = f"using {tool_name}" - return f"DM is {desc}..." - - def describe_change(tool_name: str, args: dict) -> str: """Build a compact human-readable change description from a tool call.""" if tool_name == "modify_vitals": @@ -317,147 +230,33 @@ def describe_change(tool_name: str, args: dict) -> str: return "" -# ── Changes Block Parser ──────────────────────────────────────────────────── - -def parse_changes_block(changes_block: str) -> list[dict]: - """Parse a ### Changes block into tool call dicts.""" - calls = [] - for raw_line in changes_block.split("\n"): - line = raw_line.strip() - if not line.startswith("- "): - continue - content = line[2:].strip() - - m = re.match(r"Current Health:\s*(\d+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "modify_vitals", "args": {"current_hp": m.group(1)}}) - continue - - m = re.match(r"Cash:\s*(\d+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "modify_vitals", "args": {"cash": m.group(1)}}) - continue - - m = re.match(r"Max Health:\s*(\d+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "modify_vitals", "args": {"max_hp": m.group(1)}}) - continue - - m = re.match(r"Add(?:ed)? to inventory:\s*(.+)", content, re.IGNORECASE) - if m: - for item in [i.strip() for i in m.group(1).split(",") if i.strip()]: - calls.append({"tool": "add_to_inventory", "args": {"item": item}}) - continue - - m = re.match(r"Remov(?:e|ed) from inventory:\s*(.+)", content, re.IGNORECASE) - if m: - for item in [i.strip() for i in m.group(1).split(",") if i.strip()]: - calls.append({"tool": "remove_from_inventory", "args": {"item": item}}) - continue - - m = re.match(r"Replace(?:d)? gear:\s*(.+?)\s*[→➜]\s*(.+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "replace_gear", "args": {"before": m.group(1).strip(), "after": m.group(2).strip()}}) - continue - - m = re.match(r"Note:\s*(.+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "add_note", "args": {"note": m.group(1).strip()}}) - continue - - m = re.match(r"Journal add:\s*(.+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "journal_update", "args": {"add": [i.strip() for i in m.group(1).split(",") if i.strip()]}}) - continue - - m = re.match(r"Journal done:\s*(.+)", content, re.IGNORECASE) - if m: - calls.append({"tool": "journal_update", "args": {"done": [i.strip() for i in m.group(1).split(",") if i.strip()]}}) - continue - - m = re.match(r"Looted from .+:\s*(.+)", content, re.IGNORECASE) - if m: - items_text = m.group(1).strip() - calls.append({"tool": "add_note", "args": {"note": f"Looted: {items_text}"}}) - continue - - return calls - - -# ── Extraction Functions ──────────────────────────────────────────────────── - -def extract_thoughts(text: str) -> list[str]: - pattern = r"```thought\s*\n?(.*?)```" - return re.findall(pattern, text, re.DOTALL) - - -def extract_tool_calls(text: str, *, round_num: int = 0, on_debug: callable = None) -> list[dict]: - """Extract tool calls from ```tool and ```json blocks.""" +def extract_tool_calls(text: str, on_debug: callable = None) -> list[dict]: + """Extract tool calls from ```tool blocks in LLM response.""" calls = [] seen = set() - def _try_parse(raw: str) -> dict | None: - try: - obj = json.loads(raw) - if isinstance(obj, dict) and "tool" in obj: - return obj - except json.JSONDecodeError: - pass - return None - - for m in re.finditer(r"```(?:tool|json|finalize_turn)\s*\n?", text): - fence_type = m.group(0).strip("``` \n\r") - obj = None + for m in re.finditer(r"```tool\s*\n?", text): try: decoder = json.JSONDecoder() obj, end = decoder.raw_decode(text, m.end()) except (json.JSONDecodeError, ValueError, StopIteration): - pass - - if obj is None: close = text.find("```", m.end()) if close > 0: raw = text[m.end():close].strip() + raw = re.sub(r'"(?:[^"\\]|\\.)*"', lambda x: x.group(0).replace("\n", "\\n"), raw, flags=re.DOTALL) + try: + obj = json.loads(raw) + except json.JSONDecodeError: + continue + else: + continue - def _escape_in_strings(s: str) -> str: - return re.sub(r'"(?:[^"\\]|\\.)*"', lambda x: x.group(0).replace("\n", "\\n"), s, flags=re.DOTALL) - repaired = _escape_in_strings(raw) - obj = _try_parse(repaired) + if not isinstance(obj, dict) or "tool" not in obj: + continue - if obj is not None and isinstance(obj, dict): - if fence_type == "finalize_turn": - obj = {"tool": "finalize_turn", "args": obj} - if "tool" not in obj: - obj = None - - if obj is not None: - key = (obj["tool"], json.dumps(obj.get("args", {}), sort_keys=True)) - if key not in seen: - seen.add(key) - calls.append(obj) - elif on_debug: - preview = text[m.end():m.end() + 120].replace("\n", "\\n") - on_debug("parse_error", {"round": round_num, "content": preview}) + key = (obj["tool"], json.dumps(obj.get("args", {}), sort_keys=True)) + if key not in seen: + seen.add(key) + calls.append(obj) return calls - - -def extract_final_json(text: str) -> dict | None: - pattern = r"```json\s*\n?(.*?)```" - matches = re.findall(pattern, text, re.DOTALL) - if not matches: - return None - try: - return json.loads(matches[-1].strip()) - except json.JSONDecodeError: - return None - - -def strip_tool_blocks(text: str) -> str: - """Remove ```tool, ```json, finalize_turn blocks from narrative text.""" - return re.sub( - r'```(?:tool|json|finalize_turn)\s*\n?.*?```', - '', - text, - flags=re.DOTALL, - ).strip() diff --git a/tools/engine_lib/validation.py b/tools/engine_lib/validation.py index 1ec4cd2..b1fe975 100644 --- a/tools/engine_lib/validation.py +++ b/tools/engine_lib/validation.py @@ -1,15 +1,6 @@ -#!/usr/bin/env python3 -""" -validation.py — Narrative quality validation for The Chaos engine. - -Standalone functions — no dependency on GameEngine. -""" - from __future__ import annotations import json -import re -from collections import Counter from .llm import call_llm from .paths import CHAR_PATH, WORLD_PATH @@ -29,7 +20,6 @@ or ## World {world} - ## Player Action {action} @@ -45,9 +35,6 @@ Reply with ONLY the JSON object.""" def validate_action( player_action: str, - *, - model: str | None = None, - timeout: int | None = None, on_debug: callable = None, ) -> tuple[bool, str]: """Ask the LLM whether a player action is valid given the game state. Returns (valid, reason).""" @@ -61,8 +48,6 @@ def validate_action( text = call_llm( [{"role": "user", "content": prompt}], - model=model, - timeout=timeout, max_tokens=256, temperature=0.2, label="Action validation", @@ -88,57 +73,3 @@ def validate_action( def auto_prompt(book_log: str = "") -> str: """Fallback player prompt.""" return "**What do you do?**" - - -def validate_narrative( - book_log: str, - *, - model: str | None = None, - 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, max_tokens=2, temperature=0.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 6d0c9bd..461cf3e 100755 --- a/tools/run.py +++ b/tools/run.py @@ -272,10 +272,7 @@ class ChaosTUI(App): self.call_from_thread(self._on_debug, event_type, data) try: - strategy = self.engine.config.get("llm", {}).get("strategy", "tools") - gen = (self.engine.generate_with_tools_single if strategy == "tools" - else self.engine.generate_with_tools) - result = gen( + result = self.engine.generate_turn( player_action=player_action, last_prompt=last_prompt, on_thought=on_thought, diff --git a/tools/store_turn.py b/tools/store_turn.py deleted file mode 100755 index bc6a0b1..0000000 --- a/tools/store_turn.py +++ /dev/null @@ -1,38 +0,0 @@ -#!/usr/bin/env python3 -"""Archive the current turn to the story book and clear turn temp files. - -Usage: - python3 tools/store_turn.py -""" - -from pathlib import Path -from datetime import date - -SESSION = Path(__file__).resolve().parent.parent / 'session' -BOOK = SESSION / 'book.md' -TURN_DESC = SESSION / 'turn_description.md' - -CLEAR_FILES = [TURN_DESC] - - -def main(): - description = TURN_DESC.read_text().strip() if TURN_DESC.exists() else '' - if not description: - print("Nothing to store — turn_description.md is empty.") - return - - timestamp = date.today().isoformat() - heading = f'\n\n## Turn — {timestamp}\n\n' - BOOK.parent.mkdir(parents=True, exist_ok=True) - with open(BOOK, 'a') as f: - f.write(heading + description + '\n') - - for fpath in CLEAR_FILES: - if fpath.exists(): - fpath.write_text('') - - print(f"✓ Turn stored → {BOOK}") - - -if __name__ == '__main__': - main() diff --git a/tools/test_imports.py b/tools/test_imports.py index ac49ef2..11329bc 100755 --- a/tools/test_imports.py +++ b/tools/test_imports.py @@ -17,7 +17,6 @@ MODULES = [ '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 15e8aed..221fdd0 100755 --- a/tools/test_runtime.py +++ b/tools/test_runtime.py @@ -23,16 +23,15 @@ def test_engine_import(): modules_to_test = [ ('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.models', ['TurnResult']), + ('engine_lib.prompts', ['SYSTEM_PROMPT']), ('engine_lib.config', ['load_config', 'save_config', 'get_model']), - ('engine_lib.context', ['build_system_prompt', 'build_user_message', 'build_prose_prompt']), + ('engine_lib.context', ['build_system_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', 'validate_action']), - ('engine_lib.parsing', ['parse_response', 'log_turn_details']), - ('engine_lib.strategies', ['generate_with_tools', 'generate_with_tools_single']), + ('engine_lib.llm', ['call_llm']), + ('engine_lib.validation', ['auto_prompt', 'validate_action']), + ('engine_lib.parsing', ['log_turn_details']), ('engine', ['GameEngine']), ] @@ -50,12 +49,11 @@ def test_engine_import(): else: print(f" ✓ {mod_name}.{attr} exists") - # Check that GameEngine has generate_with_tools_single import engine - if hasattr(engine.GameEngine, 'generate_with_tools_single'): - print(f"✓ engine.GameEngine.generate_with_tools_single method found") + if hasattr(engine.GameEngine, 'generate_turn'): + print(f"✓ engine.GameEngine.generate_turn method found") else: - errors.append("engine.GameEngine.generate_with_tools_single method not found") + errors.append("engine.GameEngine.generate_turn method not found") return errors