8c06fd1be7
Adds _validate_update_size() safeguard that skips writing if the LLM returns a fragment instead of the full file — warns on stderr.
915 lines
35 KiB
Python
915 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
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.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
from string import Template
|
||
from typing import Iterator, Optional
|
||
|
||
|
||
# ── Paths ──────────────────────────────────────────────────────────────────
|
||
BASE_DIR = Path(__file__).resolve().parent.parent
|
||
SESSION_DIR = BASE_DIR / 'session'
|
||
CONFIG_PATH = SESSION_DIR / 'config.json'
|
||
CHAR_PATH = SESSION_DIR / 'character.md'
|
||
WORLD_PATH = SESSION_DIR / 'world.md'
|
||
BOOK_PATH = SESSION_DIR / 'book.md'
|
||
JOURNAL_PATH = SESSION_DIR / 'journal.md'
|
||
AMBIENCE_PATH = SESSION_DIR / 'ambience.md'
|
||
LOG_DIR = SESSION_DIR / 'log'
|
||
TODAY = date.today().isoformat()
|
||
|
||
|
||
# ── Structured output ──────────────────────────────────────────────────────
|
||
@dataclass
|
||
class GenerationResult:
|
||
"""Legacy result — kept for backward compat with CLI main()."""
|
||
narrative: str
|
||
choices: list[str] = field(default_factory=list)
|
||
log_entry: Optional[str] = None
|
||
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."""
|
||
book_log: str = ""
|
||
user_prompt: str = ""
|
||
ambience: Optional[str] = None
|
||
character_updates: Optional[str] = None
|
||
world_updates: Optional[str] = None
|
||
journal_add: list[str] = field(default_factory=list)
|
||
journal_done: list[str] = field(default_factory=list)
|
||
error: Optional[str] = None
|
||
debug_info: str = ""
|
||
|
||
|
||
# ── DM System Prompt Template ──────────────────────────────────────────────
|
||
SYSTEM_PROMPT = Template("""You are the Dungeon Master for **The Chaos**, a solo card-based rules-light fantasy TTRPG. Your job is to narrate an immersive, responsive story for one player character.
|
||
|
||
## Tone & Style
|
||
- Write in **second person** ("You", "Dillion") — the player is Dillion.
|
||
- Use vivid sensory descriptions — sight, sound, smell, touch.
|
||
- Keep narration tight and cinematic. No monologues.
|
||
- Use **bold** for emphasis, *italic* for thoughts/sounds.
|
||
- NPC dialogue goes in **"quotes with bold names."**
|
||
- Never present predefined choices — the player decides freely what to do.
|
||
- Each turn should advance the story meaningfully.
|
||
|
||
## Game Rules (Quick Reference)
|
||
|
||
### Core Dice
|
||
- **Odds**: 1d6, 4+ favours character, 3- is trouble.
|
||
- **Traits**: 3d6, must roll UNDER the trait score.
|
||
- **Combat hit**: 1d6 ± mods, 4+ hits.
|
||
- **Damage**: 1d6 ± weapon mod - armour reduction.
|
||
- **Initiative**: both sides roll 1d6, higher acts first.
|
||
|
||
### Combat Flow
|
||
1. Distance: 2d6 × 10 (metres/feet)
|
||
2. Surprise: 1d6
|
||
3. Grit: 2d6 for creatures (higher = more determined)
|
||
4. Initiative: 1d6
|
||
5. Turns: state intent → roll 1d6 ± mods → 4+ success, 3- take hit
|
||
|
||
### Wounds (0 HP)
|
||
1d6: 1-2 die, 3-4 lasting wound (-1 max HP), 5-6 -1 all rolls until healed
|
||
|
||
### Roll Modifiers
|
||
Favourable +1, Risky -1, Desperate -2, Well-prepared +1, Poor visibility -1, Relevant trait +1
|
||
|
||
### Exploration
|
||
6 ten-minute watches per hour. Each meaningful action advances a watch.
|
||
|
||
## How Turns Work
|
||
|
||
Each turn follows this sequence:
|
||
1. The player's action or response is given to you.
|
||
2. Think about what happens. Read game state files, roll dice, or ask the player to roll.
|
||
3. When ready, call **finalize_turn** to complete the turn.
|
||
|
||
The **finalize_turn** tool produces all data for this turn:
|
||
- **book_log** — Narrative of what happened this turn. Appended to the story book.
|
||
- **user_prompt** — What the player sees next: describe the situation and ask what they do.
|
||
- **ambience** — One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds.
|
||
- **character_updates** — Full character sheet (ONLY if HP/cash/gear/stats changed, otherwise omit).
|
||
- **world_updates** — Full world state (ONLY if NPCs/locations/threads changed, otherwise omit).
|
||
- **journal_add** — New TODO items.
|
||
- **journal_done** — Completed TODO items.
|
||
|
||
IMPORTANT: You MUST call **finalize_turn** to end the turn. Until then you will be called again to continue thinking and gathering information.
|
||
|
||
## Available Tools
|
||
|
||
Tool calls go in their own fenced code block:
|
||
|
||
```tool
|
||
{"tool": "tool_name", "args": {...}}
|
||
```
|
||
|
||
You may also show reasoning inline:
|
||
|
||
```thought
|
||
Your reasoning here
|
||
```
|
||
|
||
Tools available:
|
||
|
||
Every tool call **must** include a `"dm_status"` string in `args` — a short, public-facing description of what the DM is doing (e.g. `"consulting the archives"`, `"examining the wound"`, `"calculating the odds"`). The player sees this in the UI. Keep it vague — never reveal what the DM is actually reading or learning.
|
||
|
||
- **read_file** — Read a game state file. `{"file": "character|world|book|log|journal", "dm_status": "..."}`
|
||
- **roll** — Auto-roll dice (outcome shown in status). `{"dice": "2d6", "modifier": "-1", "dm_status": "..."}`
|
||
- **player_roll** — Ask the player to roll physical dice. **Use when the outcome is uncertain.** `{"dice": "2d6", "reason": "why", "dm_status": "..."}`
|
||
- **finalize_turn** — **Complete the turn.** Provide all turn data as args. **Must include** `"dm_status"`.
|
||
|
||
When the player makes a choice, resolve it with the dice mechanics above. Describe the action, roll dice implicitly (describe the outcome, don't say "rolling dice"), apply damage/effects, and update state.
|
||
|
||
## Current Game State
|
||
|
||
### Character
|
||
$character
|
||
|
||
### World
|
||
$world
|
||
|
||
### Recent Log
|
||
$log
|
||
|
||
### Recent Story (last turns from the book)
|
||
$story""")
|
||
# trailing """ is intentional — the template ends here
|
||
|
||
|
||
# ── Game Engine ────────────────────────────────────────────────────────────
|
||
class GameEngine:
|
||
"""Owns the LLM interaction and game state persistence."""
|
||
|
||
def __init__(self, session_dir: str | Path = SESSION_DIR):
|
||
self.session_dir = Path(session_dir)
|
||
self.config: dict = {}
|
||
self._load_config()
|
||
|
||
# ── 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,
|
||
}
|
||
}
|
||
self._save_config()
|
||
else:
|
||
raw = CONFIG_PATH.read_text()
|
||
self.config = json.loads(raw)
|
||
# Ensure api_key is None not empty string
|
||
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")
|
||
|
||
@property
|
||
def model(self) -> str:
|
||
return self.config.get("llm", {}).get("model", "ollama/llama3.1")
|
||
|
||
@property
|
||
def api_key(self) -> str | None:
|
||
return self.config.get("llm", {}).get("api_key")
|
||
|
||
@property
|
||
def api_base(self) -> str | None:
|
||
return self.config.get("llm", {}).get("api_base")
|
||
|
||
@property
|
||
def temperature(self) -> float:
|
||
return self.config.get("llm", {}).get("temperature", 0.8)
|
||
|
||
def _set_llm_env(self) -> None:
|
||
"""Set provider-specific env vars for litellm."""
|
||
prefix = self.model.split("/")[0].upper()
|
||
import os
|
||
key = self.api_key or "sk-placeholder"
|
||
os.environ[f"{prefix}_API_KEY"] = key
|
||
if self.api_base:
|
||
os.environ[f"{prefix}_API_BASE"] = self.api_base
|
||
|
||
# ── Context Assembly ────────────────────────────────────────────────
|
||
|
||
def _read_file(self, path: Path) -> str:
|
||
return path.read_text().strip() if path.exists() else ""
|
||
|
||
def _read_recent_log(self, max_entries: int = 15) -> str:
|
||
"""Read the latest log file and return the last N entries."""
|
||
log_path = LOG_DIR / f"{TODAY}.md"
|
||
if not log_path.exists():
|
||
# Check yesterday's log
|
||
from datetime import timedelta
|
||
yesterday = (date.today() - timedelta(days=1)).isoformat()
|
||
log_path = LOG_DIR / f"{yesterday}.md"
|
||
if not log_path.exists():
|
||
return "*No recent events.*"
|
||
lines = log_path.read_text().splitlines()
|
||
entries = [l for l in lines if l.strip().startswith("- ")]
|
||
return "\n".join(entries[-max_entries:]) or "*No recent events.*"
|
||
|
||
def _read_recent_book(self, max_turns: int = 3) -> str:
|
||
"""Return the last N turns from the book as context."""
|
||
text = self._read_file(BOOK_PATH)
|
||
if not text:
|
||
return "*No prior story.*"
|
||
turns = text.split("\n## ")
|
||
recent = turns[-max_turns:]
|
||
return "\n## ".join(recent) if len(turns) > 1 else recent[0]
|
||
|
||
def build_system_prompt(self) -> str:
|
||
"""Assemble the system prompt with current game state."""
|
||
char = self._read_file(CHAR_PATH) or "*No character sheet.*"
|
||
world = self._read_file(WORLD_PATH) or "*No world state.*"
|
||
log = self._read_recent_log()
|
||
story = self._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 Action\n{player_action}")
|
||
|
||
has_existing_story = bool(
|
||
self._read_file(BOOK_PATH).strip()
|
||
) if not last_prompt else True
|
||
|
||
if not player_action and not last_prompt:
|
||
if has_existing_story:
|
||
parts.append(
|
||
"## Instructions\n"
|
||
"Continue the story from where it left off. Think, "
|
||
"gather information, then call finalize_turn."
|
||
)
|
||
else:
|
||
parts.append(
|
||
"## Instructions\n"
|
||
"Establish the opening scene. Dillion is at the "
|
||
"Splintered Tankard in the Keep. Describe the "
|
||
"setting, then call finalize_turn."
|
||
)
|
||
else:
|
||
parts.append(
|
||
"## Instructions\n"
|
||
"Describe the outcome of the player's action using game "
|
||
"mechanics where appropriate. Think, gather information, "
|
||
"then call finalize_turn to complete the turn."
|
||
)
|
||
return "\n\n".join(parts)
|
||
|
||
# ── LLM Call ────────────────────────────────────────────────────────
|
||
|
||
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.
|
||
|
||
The TUI calls this from a worker thread — see run.py.
|
||
"""
|
||
system = self.build_system_prompt()
|
||
user = self.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 API key / base if provided
|
||
self._set_llm_env()
|
||
|
||
try:
|
||
response = litellm.completion(
|
||
model=self.model,
|
||
messages=messages,
|
||
temperature=self.temperature,
|
||
stream=False,
|
||
timeout=30,
|
||
)
|
||
text = response.choices[0].message.content or ""
|
||
except Exception as e:
|
||
return GenerationResult(
|
||
narrative="",
|
||
error=f"LLM call failed: {e}",
|
||
)
|
||
|
||
return 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
|
||
)
|
||
|
||
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
|
||
|
||
self._set_llm_env()
|
||
|
||
try:
|
||
response = litellm.completion(
|
||
model=self.model,
|
||
messages=messages,
|
||
temperature=self.temperature,
|
||
stream=True,
|
||
timeout=30,
|
||
)
|
||
full_text = ""
|
||
for chunk in response:
|
||
delta = chunk.choices[0].delta.content or ""
|
||
if delta:
|
||
full_text += delta
|
||
yield full_text # partial narrative for streaming display
|
||
# Final yield is the completed text
|
||
yield full_text
|
||
except Exception as e:
|
||
yield json.dumps({"error": f"LLM call failed: {e}"})
|
||
|
||
# ── Tool Infrastructure ────────────────────────────────────────────
|
||
|
||
TOOL_REGISTRY: dict[str, dict] = {
|
||
"read_file": {
|
||
"description": "Read a game state file.",
|
||
"args": {"file": "character | world | book | log | journal"},
|
||
},
|
||
"roll": {
|
||
"description": "Roll dice and return the outcome.",
|
||
"args": {"dice": "e.g. 1d6, 2d6", "modifier": "optional +N or -N"},
|
||
},
|
||
"think": {
|
||
"description": "Internal reasoning shown in the game status bar.",
|
||
"args": {"thought": "Your reasoning."},
|
||
},
|
||
"player_roll": {
|
||
"description": "Ask the player to physically roll dice and enter the result.",
|
||
"args": {"dice": "e.g. 2d6+1", "reason": "Why the roll is needed (shown to player)"},
|
||
},
|
||
"finalize_turn": {
|
||
"description": "Complete the turn with all required data.",
|
||
"args": {
|
||
"book_log": "Narrative of what happened (appended to story book)",
|
||
"user_prompt": "What the player sees next — describe and ask what they do",
|
||
"ambience": "Optional: soundscape name",
|
||
"character_updates": "Optional: full character sheet if changed",
|
||
"world_updates": "Optional: full world state if changed",
|
||
"journal_add": "Optional: list of new TODO items",
|
||
"journal_done": "Optional: list of completed TODO items",
|
||
},
|
||
},
|
||
}
|
||
|
||
def _tool_think(self, args: dict) -> str:
|
||
"""Think tool — content is displayed via dm_status in the status bar."""
|
||
return ""
|
||
|
||
def _tool_read_file(self, args: dict) -> str:
|
||
filename = (args or {}).get("file", "")
|
||
paths = {
|
||
"character": CHAR_PATH,
|
||
"world": WORLD_PATH,
|
||
"book": BOOK_PATH,
|
||
"log": LOG_DIR / f"{TODAY}.md",
|
||
"journal": JOURNAL_PATH,
|
||
}
|
||
path = paths.get(filename)
|
||
if not path:
|
||
return f"Unknown file: {filename}. Choose from: {', '.join(paths)}"
|
||
return self._read_file(path) or f"*{filename} is empty.*"
|
||
|
||
def _tool_roll(self, args: dict) -> str:
|
||
import random
|
||
dice_str = (args or {}).get("dice", "1d6")
|
||
modifier_str = (args or {}).get("modifier", "0")
|
||
try:
|
||
count, sides = dice_str.lower().split("d")
|
||
count = int(count) if count else 1
|
||
sides = int(sides)
|
||
except (ValueError, TypeError):
|
||
return f"Invalid dice: {dice_str}. Use format like '2d6'."
|
||
mod = 0
|
||
if modifier_str:
|
||
try:
|
||
mod = int(modifier_str)
|
||
except ValueError:
|
||
pass
|
||
rolls = [random.randint(1, sides) for _ in range(count)]
|
||
total = sum(rolls) + mod
|
||
mod_str = f" {'+' if mod >= 0 else ''}{mod}" if mod != 0 else ""
|
||
return f"Roll: {dice_str}{mod_str} → [{', '.join(str(r) for r in rolls)}] = {total}"
|
||
|
||
@staticmethod
|
||
def _describe_tool_action(tool_name: str, args: dict) -> str:
|
||
"""Return a user-facing status message for a tool call.
|
||
Prefer the LLM-provided dm_status — otherwise fall back to a generic description."""
|
||
dm_status = (args or {}).get("dm_status")
|
||
if dm_status:
|
||
return f"DM is {dm_status}..."
|
||
|
||
read_descriptions = {
|
||
"character": "reading the character sheet",
|
||
"world": "consulting the world map",
|
||
"book": "reviewing the story so far",
|
||
"log": "checking the session log",
|
||
"journal": "scanning the journal",
|
||
}
|
||
if tool_name == "read_file":
|
||
file = (args or {}).get("file", "")
|
||
desc = read_descriptions.get(file, f"reading {file}")
|
||
elif tool_name == "roll":
|
||
dice = (args or {}).get("dice", "1d6")
|
||
mod = (args or {}).get("modifier")
|
||
desc = f"rolling {dice}"
|
||
if mod:
|
||
desc += f" {mod}"
|
||
elif tool_name == "player_roll":
|
||
dice = (args or {}).get("dice", "1d6")
|
||
desc = f"asking you to roll {dice}"
|
||
else:
|
||
desc = f"using {tool_name}"
|
||
return f"DM is {desc}..."
|
||
|
||
def _execute_tool(self, tool_name: str, args: dict) -> str:
|
||
fn_map = {
|
||
"read_file": self._tool_read_file,
|
||
"roll": self._tool_roll,
|
||
"think": self._tool_think,
|
||
}
|
||
fn = fn_map.get(tool_name)
|
||
if not fn:
|
||
return f"Unknown tool: {tool_name}"
|
||
try:
|
||
return fn(args)
|
||
except Exception as e:
|
||
return f"Tool error ({tool_name}): {e}"
|
||
|
||
@staticmethod
|
||
def _extract_thoughts(text: str) -> list[str]:
|
||
pattern = r"```thought\s*\n?(.*?)```"
|
||
return re.findall(pattern, text, re.DOTALL)
|
||
|
||
@staticmethod
|
||
def _extract_tool_calls(text: str) -> list[dict]:
|
||
pattern = r"```tool\s*\n?(.*?)```"
|
||
blocks = re.findall(pattern, text, re.DOTALL)
|
||
calls = []
|
||
for block in blocks:
|
||
try:
|
||
parsed = json.loads(block.strip())
|
||
if isinstance(parsed, dict) and "tool" in parsed:
|
||
calls.append(parsed)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return calls
|
||
|
||
@staticmethod
|
||
def _extract_final_json(text: str) -> dict | None:
|
||
pattern = r"```json\s*\n?(.*?)```"
|
||
matches = re.findall(pattern, text, re.DOTALL)
|
||
if not matches:
|
||
return None
|
||
try:
|
||
return json.loads(matches[-1].strip())
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
def generate_with_tools(
|
||
self,
|
||
player_action: str | None = None,
|
||
last_prompt: str | None = None,
|
||
on_thought: callable = None,
|
||
on_action: callable = None,
|
||
on_player_roll: callable = None,
|
||
) -> TurnResult:
|
||
"""
|
||
Multi-turn generation with tool-use loop.
|
||
|
||
The LLM can output ```thought blocks, call ```tool blocks, and
|
||
MUST call **finalize_turn** to complete the turn. Until then the
|
||
loop continues feeding tool results back.
|
||
|
||
`on_thought` / `on_action` may be called from a worker thread —
|
||
use call_from_thread in the TUI.
|
||
"""
|
||
system = self.build_system_prompt()
|
||
user = self.build_user_message(
|
||
player_action=player_action,
|
||
last_prompt=last_prompt,
|
||
)
|
||
|
||
messages: list[dict] = [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": user},
|
||
]
|
||
|
||
self._set_llm_env()
|
||
|
||
try:
|
||
import litellm
|
||
except ImportError:
|
||
return TurnResult(error="litellm not installed")
|
||
|
||
max_rounds = 10
|
||
debug_entries: list[str] = []
|
||
|
||
for round_idx in range(max_rounds):
|
||
round_log: list[str] = [f"── Round {round_idx + 1} ──"]
|
||
|
||
try:
|
||
response = litellm.completion(
|
||
model=self.model,
|
||
messages=messages,
|
||
temperature=self.temperature,
|
||
stream=False,
|
||
timeout=30,
|
||
)
|
||
text = response.choices[0].message.content or ""
|
||
except Exception as e:
|
||
return TurnResult(error=f"LLM call failed: {e}")
|
||
|
||
# Thoughts
|
||
thoughts = self._extract_thoughts(text)
|
||
if thoughts:
|
||
round_log.append(f" thoughts: {len(thoughts)}")
|
||
for t in thoughts:
|
||
if on_thought:
|
||
on_thought(t.strip())
|
||
|
||
# Tool calls
|
||
tool_calls = self._extract_tool_calls(text)
|
||
finalize_call: dict | None = None
|
||
other_calls: list[dict] = []
|
||
|
||
for tc in tool_calls:
|
||
if tc.get("tool") == "finalize_turn":
|
||
finalize_call = tc
|
||
else:
|
||
other_calls.append(tc)
|
||
|
||
# Log tool call summary
|
||
if tool_calls:
|
||
names = [tc.get("tool", "?") for tc in tool_calls]
|
||
round_log.append(f" tools: {', '.join(names)}")
|
||
|
||
# finalize_turn present → validate and return
|
||
if finalize_call:
|
||
args = finalize_call.get("args", {})
|
||
errs = []
|
||
if not args.get("book_log"):
|
||
errs.append("book_log is required")
|
||
if not args.get("user_prompt"):
|
||
errs.append("user_prompt is required")
|
||
if errs:
|
||
round_log.append(f" finalize_turn validation errors: {', '.join(errs)}")
|
||
debug_entries.append("\n".join(round_log))
|
||
messages.append({"role": "assistant", "content": text})
|
||
messages.append({
|
||
"role": "user",
|
||
"content": f"## Validation Error\nfinalize_turn missing: {', '.join(errs)}. Please provide all required fields and call finalize_turn again."
|
||
})
|
||
continue
|
||
return TurnResult(
|
||
book_log=args.get("book_log", ""),
|
||
user_prompt=args.get("user_prompt", ""),
|
||
ambience=args.get("ambience"),
|
||
character_updates=args.get("character_updates"),
|
||
world_updates=args.get("world_updates"),
|
||
journal_add=args.get("journal_add", []),
|
||
journal_done=args.get("journal_done", []),
|
||
)
|
||
|
||
# Execute other tools
|
||
if other_calls:
|
||
results = []
|
||
for tc in other_calls:
|
||
name = tc.get("tool", "?")
|
||
args = tc.get("args", {})
|
||
|
||
# dm_status is required on every tool call
|
||
if not args.get("dm_status"):
|
||
err_msg = (
|
||
f"**Validation Error:** Tool `{name}` missing required `dm_status`. "
|
||
f"Describe what the DM is doing (e.g. "
|
||
f'`"dm_status": "consulting the archives"`). Please retry.'
|
||
)
|
||
results.append(err_msg)
|
||
round_log.append(f" {name}: MISSING dm_status")
|
||
continue
|
||
|
||
if on_action:
|
||
on_action(self._describe_tool_action(name, args))
|
||
if name == "player_roll" and on_player_roll:
|
||
dice = args.get("dice", "1d6")
|
||
reason = args.get("reason", "a check")
|
||
roll_val = on_player_roll(dice, reason)
|
||
result = f"Player rolled {dice} for '{reason}': {roll_val}"
|
||
else:
|
||
result = self._execute_tool(name, args)
|
||
results.append(f"**Tool:** {name}\n**Args:** {json.dumps(args)}\n**Result:** {result}")
|
||
round_log.append(f" {name}: OK")
|
||
messages.append({"role": "assistant", "content": text})
|
||
messages.append({
|
||
"role": "user",
|
||
"content": "## Tool Results\n\n" + "\n\n".join(results),
|
||
})
|
||
debug_entries.append("\n".join(round_log))
|
||
continue
|
||
|
||
# No tools, no finalize → remind LLM
|
||
round_log.append(" no tool calls — prompted to use tools")
|
||
debug_entries.append("\n".join(round_log))
|
||
messages.append({"role": "assistant", "content": text})
|
||
messages.append({
|
||
"role": "user",
|
||
"content": "## Instructions\nUse tools to gather information or call **finalize_turn** to complete the turn."
|
||
})
|
||
|
||
debug_text = "\n\n".join(debug_entries)
|
||
return TurnResult(
|
||
error=f"Turn loop exceeded max rounds ({max_rounds}). Below is a debug log of what the LLM did each round:\n\n{debug_text}",
|
||
debug_info=debug_text,
|
||
)
|
||
|
||
# ── Response Parsing ────────────────────────────────────────────────
|
||
|
||
@staticmethod
|
||
def parse_response(text: str) -> GenerationResult:
|
||
"""
|
||
Parse a full LLM response into a GenerationResult.
|
||
Extracts the JSON block and splits narrative from it.
|
||
"""
|
||
# Check for error JSON
|
||
if text.startswith('{"error":'):
|
||
try:
|
||
err = json.loads(text).get("error", "Unknown error")
|
||
except json.JSONDecodeError:
|
||
err = "Unknown error"
|
||
return GenerationResult(narrative="", error=err)
|
||
|
||
# Try to find a ```json ... ``` block
|
||
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 = narrative.strip()
|
||
try:
|
||
data = json.loads(json_str)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
else:
|
||
# Fallback: maybe the entire response is JSON (no fence)
|
||
text_stripped = text.strip()
|
||
if text_stripped.startswith("{") and text_stripped.endswith("}"):
|
||
try:
|
||
data = json.loads(text_stripped)
|
||
narrative = data.get("narrative", "")
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
return GenerationResult(
|
||
narrative=narrative or text,
|
||
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", []),
|
||
)
|
||
|
||
# ── State Persistence ───────────────────────────────────────────────
|
||
|
||
def _validate_update_size(self, name: str, new_content: str, path: Path) -> bool:
|
||
"""Reject updates that are more than 30% shorter than the existing file
|
||
— likely the LLM pasted a fragment instead of the full state."""
|
||
if not path.exists():
|
||
return True
|
||
old = path.read_text().strip()
|
||
if not old:
|
||
return True
|
||
ratio = len(new_content) / len(old)
|
||
if ratio < 0.7:
|
||
import sys
|
||
print(
|
||
f"WARNING: {name} update rejected ({ratio:.0%} of original size "
|
||
f"= {len(new_content)} vs {len(old)} chars) — likely a partial paste.",
|
||
file=sys.stderr,
|
||
)
|
||
return False
|
||
return True
|
||
|
||
def apply_state(self, result: TurnResult) -> None:
|
||
"""Write state changes from a TurnResult to disk."""
|
||
|
||
if result.character_updates and self._validate_update_size(
|
||
"character", result.character_updates, CHAR_PATH
|
||
):
|
||
CHAR_PATH.write_text(result.character_updates.strip() + "\n")
|
||
|
||
if result.world_updates and self._validate_update_size(
|
||
"world", result.world_updates, WORLD_PATH
|
||
):
|
||
WORLD_PATH.write_text(result.world_updates.strip() + "\n")
|
||
|
||
if result.ambience:
|
||
AMBIENCE_PATH.write_text(result.ambience.strip().lower() + "\n")
|
||
|
||
if result.journal_add or result.journal_done:
|
||
self._update_journal(
|
||
add=result.journal_add, done=result.journal_done
|
||
)
|
||
|
||
def archive_turn(self, narrative: str) -> None:
|
||
"""Append the narrative as a new turn in book.md."""
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
heading = f"\n\n## Turn — {timestamp}\n\n"
|
||
BOOK_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(BOOK_PATH, "a") as f:
|
||
f.write(heading + narrative.strip() + "\n")
|
||
|
||
def append_log(self, entry: str) -> None:
|
||
"""Append a log entry to today's log file."""
|
||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
log_path = LOG_DIR / f"{TODAY}.md"
|
||
if not log_path.exists():
|
||
log_path.write_text(f"# Session Log — {TODAY}\n\n")
|
||
with open(log_path, "a") as f:
|
||
f.write(entry.strip() + "\n")
|
||
|
||
def _update_journal(
|
||
self, add: list[str] | None = None, done: list[str] | None = None
|
||
) -> None:
|
||
"""Add or complete TODO items in journal.md."""
|
||
if not JOURNAL_PATH.exists():
|
||
JOURNAL_PATH.write_text("# Journal\n\n## TODO\n\n## DONE\n\n")
|
||
lines = JOURNAL_PATH.read_text().splitlines()
|
||
new_lines = []
|
||
in_todo = False
|
||
in_done = False
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if stripped.startswith("## TODO"):
|
||
in_todo = True
|
||
in_done = False
|
||
elif stripped.startswith("## DONE"):
|
||
in_todo = False
|
||
in_done = True
|
||
new_lines.append(line)
|
||
|
||
# Find insertion points
|
||
todo_idx = None
|
||
done_idx = None
|
||
for i, line in enumerate(lines):
|
||
stripped = line.strip()
|
||
if stripped == "## TODO":
|
||
todo_idx = i
|
||
elif stripped == "## DONE":
|
||
done_idx = i
|
||
|
||
if done:
|
||
for item in done:
|
||
# Remove from TODO if present
|
||
new_lines = [
|
||
l for l in new_lines
|
||
if l.strip().lstrip("- ").lstrip("☐ ") != item
|
||
]
|
||
# Find DONE section and add
|
||
if done_idx is not None:
|
||
done_entry = f"- {item}"
|
||
if done_idx + 1 < len(new_lines):
|
||
new_lines.insert(done_idx + 1, done_entry)
|
||
else:
|
||
new_lines.append(done_entry)
|
||
|
||
if add:
|
||
for item in add:
|
||
entry = f"- {item}"
|
||
if entry not in new_lines:
|
||
if todo_idx is not None:
|
||
new_lines.insert(todo_idx + 1, entry)
|
||
else:
|
||
new_lines.append(entry)
|
||
|
||
JOURNAL_PATH.write_text("\n".join(new_lines) + "\n")
|
||
|
||
|
||
# ── CLI entry point (for testing) ─────────────────────────────────────────
|
||
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")
|
||
parser.add_argument("--last", "-l", help="Last narrative text")
|
||
args = parser.parse_args()
|
||
|
||
engine = GameEngine()
|
||
result = engine.generate(
|
||
player_action=args.action,
|
||
last_narrative=args.last,
|
||
)
|
||
|
||
if result.error:
|
||
print(f"ERROR: {result.error}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
print(result.narrative)
|
||
if result.choices:
|
||
print("\n--- Choices ---")
|
||
for c in result.choices:
|
||
print(f" [{c}]")
|
||
if result.log_entry:
|
||
print(f"\n[Log] {result.log_entry}")
|
||
if result.ambience:
|
||
print(f"[Ambience] {result.ambience}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|