from __future__ import annotations import json import random import re import sys from datetime import datetime from difflib import SequenceMatcher from pathlib import Path from engine_lib.models import TurnResult, END_MARKER from engine_lib import config from engine_lib.context import build_system_prompt from engine_lib.validation import validate_turn 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 from engine_lib.paths import CHARACTER_CREATION_PATH, RULES_INJECTION_PATH class GameEngine: REQUIRED_TOOL_ARGS: dict[str, list[str]] = { "add_to_inventory": ["item"], "remove_from_inventory": ["item"], "replace_gear": ["before", "after"], "add_note": ["note"], "replace_note": ["before", "after"], "world_update": ["content"], "journal_update": ["add", "done"], } def __init__(self, session_dir: str | Path | None = None): self.config = config.load_config() def _check_required_tool_args(self, state_changes: list[dict]) -> str: """Check that all state-changing tool calls have required args. Returns empty string if OK, or a description of what's missing.""" missing = [] for tc in state_changes: name = tc.get("tool", "") req = self.REQUIRED_TOOL_ARGS.get(name) if not req: continue args = tc.get("args") or {k: v for k, v in tc.items() if k != "tool"} if name == "journal_update": if not args.get("add") and not args.get("done"): missing.append(f"{name}: needs at least one of `add` or `done`") continue for arg in req: val = args.get(arg) if val is None or (isinstance(val, str) and not val.strip()): missing.append(f"{name}: missing required `{arg}`") return "; ".join(missing) def generate_turn( self, player_action: str | None = None, recent_narrative: str | None = None, on_thought: callable = None, on_action: callable = None, ) -> TurnResult: 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}") if recent_narrative is None: recent_narrative = state.read_recent_book(2) session_log = state.read_recent_log() 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("DM is preparing a response") is_new_game = not player_action and not recent_narrative system = build_system_prompt(recent_narrative=recent_narrative, recent_log=session_log) if is_new_game: cc = state.read_file(CHARACTER_CREATION_PATH) if cc: system += f"\n\n## Character Creation Reference\n{cc}" state.append_llm_log(f"\n[NEW GAME] injected character_creation.md ({len(cc)} chars)") is_meta = bool(player_action and player_action.strip().startswith(">")) base_parts = [] if player_action: base_parts.append(f"## Player's Request\n{player_action}") if is_meta: base_parts.append( "## Instructions\n" "The player's message starts with `>` — this is a meta out-of-character question to the DM. " "Do NOT advance the story. Respond as the DM in meta language, starting the response with `>`. " "Use the `narrative` tool to output your meta response. Do NOT call any other tools (no journal_update, no finalize_turn, no rolls, no state changes)." ) elif is_new_game: base_parts.append( "## Instructions\n" "This is a new story. Welcome the player and guide them through the game setup." ) else: base_parts.append( "## Instructions\n" "Advance the story based on the player's request. " "All state is shown above — write the outcome directly." ) if not is_meta: base_parts.append(f"\n*A die is cast: **{die_roll}** (1d6).*") base_user = "\n\n".join(base_parts) MAX_RETRIES = 2 tool_calls = [] book_log = "" ambience = None errors: list[str] = [] changes: list[str] = [] start_time = datetime.now() total_attempts = 0 prev_raw = "" _ = None # placeholder for attempt in range(MAX_RETRIES + 1): total_attempts = attempt + 1 user = base_user if attempt > 0: user += f"\n\n---\n\n## Your Previous Response\n\n```\n{prev_raw}\n```\n\n---\n\n## Feedback\n{feedback}" state.append_llm_log(f"\n[TOOL] Attempt {attempt + 1}/{MAX_RETRIES + 1} — {len(system)} chars system, {len(user)} chars user") text = call_llm( [{"role": "system", "content": system}, {"role": "user", "content": user}], label="Turn generation", ) if not text or not text.strip(): if attempt < MAX_RETRIES: feedback = f"Your response was empty. Generate a complete turn with narrative and state changes." state.append_llm_log("\n[RETRY] empty response") if on_action: on_action("DM is weaving the tale...") continue return TurnResult(error="LLM returned empty response after retries") raw = text.strip() state.append_llm_log(f"\n[TOOL] got {len(raw)} chars in {(datetime.now() - start_time).total_seconds() * 1000:.1f}ms") prev_raw = raw tool_calls = extract_tool_calls(raw) if not tool_calls: state.append_llm_log("\n[TOOL] no tool blocks found") # First pass — extract narrative + identify state changes (don't execute yet) book_log = "" ambience = None log_entry = None meta_log = "" state_changes: list[dict] = [] for tc in tool_calls: name = tc.get("tool", "") args = tc.get("args") or {k: v for k, v in tc.items() if k != "tool"} if name == "narrative": text = args.get("text", "") or "" if text: book_log = (book_log + "\n\n" + text) if book_log else text elif name == "finalize_turn": raw = (args.get("ambience") or "").strip().lower() if raw: valid = state.get_valid_ambiences() if raw in valid: ambience = raw else: state.append_llm_log(f"\n[WARN] invalid ambience '{raw}'") ambience = None if args.get("log_entry"): log_entry = args["log_entry"] if args.get("meta_log"): meta_log = args["meta_log"] elif name == "read_rules": cat = args.get("category", "mechanics") result = execute_tool("read_rules", {"category": cat}) state.append_llm_log(f"\n[READ RULES] loaded {len(result)} chars") RULES_INJECTION_PATH.parent.mkdir(parents=True, exist_ok=True) RULES_INJECTION_PATH.write_text(result) else: state_changes.append(tc) # Required args check — reject if any state-changing tool is missing required arguments missing_args = self._check_required_tool_args(state_changes) if missing_args: state.append_llm_log(f"\n[TURN MISSING ARGS] {missing_args}") if attempt < MAX_RETRIES: feedback = f"The following tool calls are missing required arguments: {missing_args}. Include all required fields for each tool and regenerate." state.append_llm_log(f"\n[TURN REGENERATE] (missing args) attempt {attempt + 2}") if on_action: on_action("DM is consulting the fates...") continue state.append_llm_log(f"\n[TURN MISSING ARGS EXCEEDED] accepting despite missing args") # Meta check — reject if state changes produced for a meta action if is_meta and state_changes: state.append_llm_log(f"\n[TURN META REJECTED] state changes not allowed for meta action") if attempt < MAX_RETRIES: feedback = f"This is a meta action. Do NOT call any state-changing tools. Respond only with meta text (starting with `>`) and no tool calls beyond a finalize_turn." state.append_llm_log(f"\n[TURN REGENERATE] (meta) attempt {attempt + 2}") if on_action: on_action("DM is consulting the fates...") continue state.append_llm_log(f"\n[TURN META EXCEEDED] accepting despite state changes") # Narrative check — reject if finalized with log_entry but no narrative if not is_meta and log_entry and not book_log: state.append_llm_log(f"\n[TURN NO NARRATIVE] finalized with log_entry but no narrative") if attempt < MAX_RETRIES: feedback = f"You called finalize_turn with a log_entry but produced no narrative. Every turn must include a `narrative` tool block with the story. Regenerate with both narrative and log_entry." state.append_llm_log(f"\n[TURN REGENERATE] (no narrative) attempt {attempt + 2}") if on_action: on_action("DM is weaving the tale...") continue state.append_llm_log(f"\n[TURN NO NARRATIVE EXCEEDED] accepting despite missing narrative") # Duplicate check — reject if narrative is 80%+ similar to last book entry if not is_meta and book_log: prev = state.read_recent_book(1) if prev and prev not in ("*No prior story.*",): prev_text = re.sub(r"^## Turn \d+\n\n", "", prev, flags=re.MULTILINE).strip() ratio = SequenceMatcher(None, book_log, prev_text).ratio() if ratio >= 0.8: state.append_llm_log(f"\n[TURN DUPLICATE] {ratio:.0%} match with previous turn") if attempt < MAX_RETRIES: feedback = f"The narrative is nearly identical to the previous turn. Generate something new and different." state.append_llm_log(f"\n[TURN REGENERATE] (duplicate) attempt {attempt + 2}") if on_action: on_action("DM is weaving the tale...") continue state.append_llm_log(f"\n[TURN DUPLICATE EXCEEDED] cannot generate unique narrative") return TurnResult( book_log="", log_entry="Your action was rejected — could not generate a unique narrative.", user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\nengine failed to produce unique narrative*", ) # Validate the generated turn if on_action: on_action("DM is validating the response") if player_action and book_log: valid, reason, action = validate_turn( player_action, narrative=book_log, log_entry=log_entry or "", changes=state_changes, story=recent_narrative, log=session_log, meta=is_meta, ) if valid: state.append_llm_log(f"\n[TURN VALID] {reason}") elif reason == "Unrecognized": if attempt < MAX_RETRIES: feedback = f"The validation system could not process the previous turn. Please regenerate." state.append_llm_log(f"\n[TURN REGENERATE] (unrecognized) attempt {attempt + 2}") if on_action: on_action("DM is consulting the fates...") continue state.append_llm_log(f"\n[TURN UNRECOGNIZED] cannot validate turn") return TurnResult( book_log="", log_entry="Your action was rejected — cannot validate turn.", user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\ncannot validate turn*", ) elif action == "reject": state.append_llm_log(f"\n[TURN REJECTED] {reason}") return TurnResult( book_log="", log_entry=f"Your action was rejected — {reason}.", user_prompt=f"*Your action:\n\t\"{player_action}\"\nwas rejected:\n\t{reason}*", ) elif action == "regenerate" and attempt < MAX_RETRIES: validator_tool = json.dumps({"tool": "validate", "args": {"valid": False, "reason": reason, "action": "regenerate"}}) feedback = f"The validation tool returned:\n```tool\n{validator_tool}\n```\n\nPlease regenerate the turn addressing the issues above. Keep the same player action but fix the problems described." state.append_llm_log(f"\n[TURN REGENERATE] attempt {attempt + 2}: {reason}") if on_action: on_action("DM is searching for inspiration...") continue else: state.append_llm_log(f"\n[TURN REGENERATE EXCEEDED] accepting despite: {reason}") else: state.append_llm_log("\n[TURN SKIP VALIDATION] no player action or no narrative") # Accept this turn — execute all tool calls break if is_meta: tool_calls = [tc for tc in tool_calls if tc.get("tool") == "narrative"] # Second pass — execute all tool calls extr_start = datetime.now() for tc in tool_calls: name = tc.get("tool", "") args = tc.get("args") or {k: v for k, v in tc.items() if k != "tool"} if name in ("narrative", "read_rules"): pass else: result = execute_tool(name, args) if name not in ("narrative", "finalize_turn", "player_roll", "read_rules"): 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 not log_entry and 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 game_over = END_MARKER in book_log if on_action: on_action("Turn complete") log_turn_details( player_action=player_action or "", last_prompt=recent_narrative 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, meta_log=meta_log, ) return TurnResult( book_log=book_log, log_entry=log_entry, ambience=ambience, debug_info="; ".join(errors) if errors else "", changes=changes, is_meta=is_meta, game_over=game_over, meta_log=meta_log, ) def main(): """Generate a turn from the command line (debug/testing).""" import argparse parser = argparse.ArgumentParser(description="The Chaos Game Engine (CLI)") parser.add_argument("--action", "-a", help="Player action text") args = parser.parse_args() engine = GameEngine() result = engine.generate_turn( player_action=args.action, ) if result.error: print(f"ERROR: {result.error}", file=sys.stderr) sys.exit(1) print(result.book_log) if result.user_prompt: print(f"\n{result.user_prompt}") if result.log_entry: print(f"\n[Log] {result.log_entry}") if result.ambience: print(f"[Ambience] {result.ambience}") if __name__ == "__main__": main()