dejvino/splinter-keep
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup and simplification

Browse Source
This commit is contained in:
Dejvino 2026-06-30 21:18:35 +02:00
parent 66da60225a
commit 6229e2e8c4
12 changed files with 235 additions and 1327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

266
tools/engine.py
View File

@ -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,
on_debug=on_debug,
model=self.model,
temperature=self.temperature,
timeout=self.timeout,
max_tokens=self.max_tokens,
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(""),
)
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,
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,
)
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,
)
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,
)

60
tools/engine_lib/context.py
View File

@ -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)

21
tools/engine_lib/models.py
View File

@ -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

63
tools/engine_lib/parsing.py
View File

@ -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}")

105
tools/engine_lib/prompts.py
View File

@ -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: <item>` and any other relevant change lines (e.g. `- Current Health: <new HP>`). 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""")

683
tools/engine_lib/strategies.py
View File

@ -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,
)

225
tools/engine_lib/tools_handler.py
View File

@ -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})
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()

69
tools/engine_lib/validation.py
View File

@ -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, ""

5
tools/run.py
View File

@ -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,

38
tools/store_turn.py
View File

@ -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()

1
tools/test_imports.py
View File

@ -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():

20
tools/test_runtime.py
View File

@ -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