context-bounded tool loop, debug pane, ambience mute, finalize_turn fence fix

This commit is contained in:
Dejvino 2026-06-25 21:21:41 +02:00
parent 326c8b7ba8
commit d78aad6ce4
4 changed files with 439 additions and 61 deletions

1
.gitignore vendored
View File

@ -2,3 +2,4 @@ __pycache__/
*.pyc *.pyc
.env .env
session/audio/ session/audio/
llm.log

1
session/last_prompt.md Normal file
View File

@ -0,0 +1 @@
What do you do?

View File

@ -29,6 +29,7 @@ BOOK_PATH = SESSION_DIR / 'book.md'
JOURNAL_PATH = SESSION_DIR / 'journal.md' JOURNAL_PATH = SESSION_DIR / 'journal.md'
AMBIENCE_PATH = SESSION_DIR / 'ambience.md' AMBIENCE_PATH = SESSION_DIR / 'ambience.md'
LOG_DIR = SESSION_DIR / 'log' LOG_DIR = SESSION_DIR / 'log'
LLM_LOG_PATH = SESSION_DIR / 'llm.log'
TODAY = date.today().isoformat() TODAY = date.today().isoformat()
@ -53,6 +54,7 @@ class TurnResult:
book_log: str = "" book_log: str = ""
user_prompt: str = "" user_prompt: str = ""
ambience: Optional[str] = None ambience: Optional[str] = None
log_entry: Optional[str] = None
error: Optional[str] = None error: Optional[str] = None
debug_info: str = "" debug_info: str = ""
@ -67,7 +69,8 @@ SYSTEM_PROMPT = Template("""You are the Dungeon Master for **The Chaos**, a solo
- Use **bold** for emphasis, *italic* for thoughts/sounds. - Use **bold** for emphasis, *italic* for thoughts/sounds.
- NPC dialogue goes in **"quotes with bold names."** - NPC dialogue goes in **"quotes with bold names."**
- Never present predefined choices the player decides freely what to do. - Never present predefined choices the player decides freely what to do.
- Each turn should advance the story meaningfully. - **Stick to the player's intent.** Don't invent your own actions for the player unless forced by environment or circumstance (e.g., they trigger a trap, an NPC reacts, etc.).
- **Keep turns short** each turn covers a single action or brief exchange, not a full scene. Advance the story one step at a time.
## Game Rules (Quick Reference) ## Game Rules (Quick Reference)
@ -98,13 +101,22 @@ Favourable +1, Risky -1, Desperate -2, Well-prepared +1, Poor visibility -1, Rel
Each turn follows this sequence: Each turn follows this sequence:
1. The player's action or response is given to you. 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. 2. Think, read files, roll dice, or ask the player to roll any number of steps.
3. When ready, call **finalize_turn** to complete the turn. 3. **You MUST call `finalize_turn` to end the turn.** There is no other way to complete a turn. The loop will keep calling you until you do.
The **finalize_turn** tool produces all data for this turn: The **finalize_turn** tool produces all data for this turn:
- **book_log** Narrative of what happened this turn. Appended to the story book. - **book_log** `[Required]` **Everything that happens this turn, narrated in full.** This is appended to the story book and forms the permanent record of the adventure. Include sensory details, dialogue, outcomes the whole scene.
- **user_prompt** What the player sees next: describe the situation and ask what they do. - **user_prompt** `[Required]` **Short prompt for the player only, NOT recorded in the book.** Ask what they do next. 1-3 sentences. Don't put important narrative details here — they belong in `book_log`.
- **ambience** One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds. - **log_entry** `[Optional]` One-sentence summary of what happened (action + outcome). Keep it tight.
- **ambience** `[Optional]` One of: silence, calm, combat, dungeon, forest, tavern, tension, town, wilds.
### How the Loop Works
Each round the system reads your ````tool` blocks, executes them, and feeds back the results. This repeats until you call `finalize_turn`. If you call tools but never call `finalize_turn`, the loop runs until it hits the round limit and the turn fails with an error.
So: call `finalize_turn` when the player needs to see the outcome and make their next decision.
**Important: Do not mix get tools with finalize_turn.** If you call `read_file`, `character_get`, `world_get`, or `journal_get` in a round, you are still gathering information do NOT also call `finalize_turn` in that same round. Gather first, then finalize in a separate round.
### Journal & Quest Tracking ### Journal & Quest Tracking
@ -126,14 +138,14 @@ To read or update state files, use the dedicated tools:
- **`character_get`** / **`character_update`** Read or replace the full character sheet. ONLY update when HP/cash/gear/stats change. - **`character_get`** / **`character_update`** Read or replace the full character sheet. ONLY update when HP/cash/gear/stats change.
- **`world_get`** / **`world_update`** Read or replace the full world state. ONLY update when NPCs/locations/threads change. - **`world_get`** / **`world_update`** Read or replace the full world state. ONLY update when NPCs/locations/threads change.
IMPORTANT: You MUST call **finalize_turn** to end the turn. Until then you will be called again to continue thinking and gathering information. **IMPORTANT: `finalize_turn` is mandatory.** Every turn ends with `finalize_turn`. If you don't call it, the loop will keep feeding you tool results until it hits the round limit and the turn fails. See "How the Loop Works" above.
## Available Tools ## Available Tools
Tool calls go in their own fenced code block: Tool calls go in their own fenced code block (one call per block):
```tool ```tool
{"tool": "tool_name", "args": {...}} {"tool": "read_file", "args": {"file": "character", "dm_status": "Checking Dillion's stats."}}
``` ```
You may also show reasoning inline: You may also show reasoning inline:
@ -146,16 +158,40 @@ 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. 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": "..."}` Tool reference (`[R]` = required, `[O]` = optional):
- **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": "..."}` - **read_file** Read a game state file.
- **character_get** Read the full character sheet. `{"dm_status": "..."}` `[R] file`: "character" | "world" | "book" | "log" | "journal"
- **character_update** Replace the character sheet (full content). `{"content": "...", "dm_status": "..."}` `[R] dm_status`: "..."
- **world_get** Read the full world state. `{"dm_status": "..."}` - **roll** Auto-roll dice (outcome shown in status).
- **world_update** Replace the world state (full content). `{"content": "...", "dm_status": "..."}` `[O] dice`: "2d6" (default "1d6")
- **journal_get** Read the journal (TODO / DONE). `{"dm_status": "..."}` `[O] modifier`: "-1" (default "0")
- **journal_update** Add or complete journal entries. `{"add": [...], "done": [...], "dm_status": "..."}` `[R] dm_status`: "..."
- **finalize_turn** **Complete the turn.** Provide all turn data as args. - **player_roll** Ask the player to roll physical dice. Use when the outcome is uncertain.
`[O] dice`: "2d6" (default "1d6")
`[O] reason`: "why the roll matters"
`[R] dm_status`: "..."
- **character_get** Read the full character sheet.
`[R] dm_status`: "..."
- **character_update** Replace the full character sheet.
`[R] content`: "full character sheet markdown"
`[R] dm_status`: "..."
- **world_get** Read the full world state.
`[R] dm_status`: "..."
- **world_update** Replace the full world state.
`[R] content`: "full world state markdown"
`[R] dm_status`: "..."
- **journal_get** Read the journal (TODO / DONE).
`[R] dm_status`: "..."
- **journal_update** Add or complete journal entries.
`[O] add`: ["new todo item", ...]
`[O] done`: ["completed item", ...]
`[R] dm_status`: "..."
- **finalize_turn** **REQUIRED to end the turn.** The loop will NOT stop without it. Call this ALONE do not mix with get tools.
`[R] book_log`: "full narrative of what happened this turn — appended to story book (permanent record)"
`[R] user_prompt`: "short prompt for the player — NOT recorded, 1-3 sentences"
`[O] log_entry`: "one-sentence summary (action + outcome)"
`[O] ambience`: "soundscape name: silence|calm|combat|dungeon|forest|tavern|tension|town|wilds"
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. 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.
@ -306,21 +342,24 @@ class GameEngine:
parts.append( parts.append(
"## Instructions\n" "## Instructions\n"
"Continue the story from where it left off. Think, " "Continue the story from where it left off. Think, "
"gather information, then call finalize_turn." "gather information, then call finalize_turn.\n"
"Put each tool call in its own ```tool block."
) )
else: else:
parts.append( parts.append(
"## Instructions\n" "## Instructions\n"
"Establish the opening scene. Dillion is at the " "Establish the opening scene. Dillion is at the "
"Splintered Tankard in the Keep. Describe the " "Splintered Tankard in the Keep. Describe the "
"setting, then call finalize_turn." "setting, then call finalize_turn.\n"
"Put each tool call in its own ```tool block."
) )
else: else:
parts.append( parts.append(
"## Instructions\n" "## Instructions\n"
"Describe the outcome of the player's action using game " "Describe the outcome of the player's action using game "
"mechanics where appropriate. Think, gather information, " "mechanics where appropriate. Think, gather information, "
"then call finalize_turn to complete the turn." "then call finalize_turn to complete the turn.\n"
"Put each tool call in its own ```tool block."
) )
return "\n\n".join(parts) return "\n\n".join(parts)
@ -469,11 +508,12 @@ class GameEngine:
"args": {"add": "Optional: list of new TODO items", "done": "Optional: list of completed items"}, "args": {"add": "Optional: list of new TODO items", "done": "Optional: list of completed items"},
}, },
"finalize_turn": { "finalize_turn": {
"description": "Complete the turn with all required data.", "description": "Complete the turn.",
"args": { "args": {
"book_log": "Narrative of what happened (appended to story book)", "book_log": "[Required] Full narrative — appended to story book (permanent record)",
"user_prompt": "What the player sees next — describe and ask what they do", "user_prompt": "[Required] Short prompt for player — NOT recorded, 1-3 sentences",
"ambience": "Optional: soundscape name", "log_entry": "[Optional] One-sentence summary",
"ambience": "[Optional] Soundscape name",
}, },
}, },
} }
@ -620,17 +660,61 @@ class GameEngine:
return re.findall(pattern, text, re.DOTALL) return re.findall(pattern, text, re.DOTALL)
@staticmethod @staticmethod
def _extract_tool_calls(text: str) -> list[dict]: def _extract_tool_calls(text: str, *, round_num: int = 0, on_debug: callable = None) -> list[dict]:
pattern = r"```tool\s*\n?(.*?)```" """Extract tool calls from ```tool and ```json blocks.
blocks = re.findall(pattern, text, re.DOTALL)
Uses json.JSONDecoder.raw_decode for strict parsing; falls back to
heuristics if the LLM produces unescaped newlines in string values.
"""
calls = [] calls = []
for block in blocks: seen = set()
def _try_parse(raw: str) -> dict | None:
try: try:
parsed = json.loads(block.strip()) obj = json.loads(raw)
if isinstance(parsed, dict) and "tool" in parsed: if isinstance(obj, dict) and "tool" in obj:
calls.append(parsed) return obj
except json.JSONDecodeError: except json.JSONDecodeError:
pass pass
return None
for m in re.finditer(r"```(?:tool|json|finalize_turn)\s*\n?", text):
fence_type = m.group(0).strip("``` \n\r")
# 1) Strict: raw_decode from where the JSON should start
obj = None
try:
decoder = json.JSONDecoder()
obj, end = decoder.raw_decode(text, m.end())
except (json.JSONDecodeError, ValueError, StopIteration):
pass
if obj is None:
# 2) Fallback: find closing backticks and repair unescaped newlines in strings
close = text.find("```", m.end())
if close > 0:
raw = text[m.end():close].strip()
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 obj is not None and isinstance(obj, dict):
# Normalize: fence type "finalize_turn" means the JSON is the args directly
if fence_type == "finalize_turn":
obj = {"tool": "finalize_turn", "args": obj}
# If JSON has a "tool" key, keep as-is
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 return calls
@staticmethod @staticmethod
@ -651,6 +735,7 @@ class GameEngine:
on_thought: callable = None, on_thought: callable = None,
on_action: callable = None, on_action: callable = None,
on_player_roll: callable = None, on_player_roll: callable = None,
on_debug: callable = None,
) -> TurnResult: ) -> TurnResult:
""" """
Multi-turn generation with tool-use loop. Multi-turn generation with tool-use loop.
@ -659,7 +744,7 @@ class GameEngine:
MUST call **finalize_turn** to complete the turn. Until then the MUST call **finalize_turn** to complete the turn. Until then the
loop continues feeding tool results back. loop continues feeding tool results back.
`on_thought` / `on_action` may be called from a worker thread `on_thought` / `on_action` / `on_debug` may be called from a worker thread
use call_from_thread in the TUI. use call_from_thread in the TUI.
""" """
system = self.build_system_prompt() system = self.build_system_prompt()
@ -682,9 +767,22 @@ class GameEngine:
max_rounds = 10 max_rounds = 10
debug_entries: list[str] = [] debug_entries: list[str] = []
attempt = 0
round_used = 0
reminder_count = 0
for round_idx in range(max_rounds): from datetime import datetime
round_log: list[str] = [f"── Round {round_idx + 1} ──"] self._append_llm_log(f"\n{'='*60}")
self._append_llm_log(f"=== Turn — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ===")
self._append_llm_log(f"{'='*60}")
if player_action:
self._append_llm_log(f"Player: {player_action}")
elif last_prompt:
self._append_llm_log(f"Resume from: {last_prompt[:120]}")
while round_used < max_rounds:
attempt += 1
round_log: list[str] = [f"── Attempt {attempt} (round {round_used + 1}/{max_rounds}) ──"]
try: try:
response = litellm.completion( response = litellm.completion(
@ -695,9 +793,18 @@ class GameEngine:
timeout=30, timeout=30,
) )
text = response.choices[0].message.content or "" text = response.choices[0].message.content or ""
self._append_llm_log(
f"\n--- Attempt {attempt} ---\n{text}"
)
except Exception as e: except Exception as e:
self._append_llm_log(f"\n--- LLM ERROR (attempt {attempt}) ---\n{e}")
if on_debug:
on_debug("llm_error", {"error": str(e)})
return TurnResult(error=f"LLM call failed: {e}") return TurnResult(error=f"LLM call failed: {e}")
if on_debug:
on_debug("llm_response", {"round": attempt, "text": text})
# Thoughts # Thoughts
thoughts = self._extract_thoughts(text) thoughts = self._extract_thoughts(text)
if thoughts: if thoughts:
@ -705,9 +812,15 @@ class GameEngine:
for t in thoughts: for t in thoughts:
if on_thought: if on_thought:
on_thought(t.strip()) on_thought(t.strip())
if on_debug:
on_debug("thought", {"round": attempt, "text": t.strip()})
# Tool calls # Tool calls
tool_calls = self._extract_tool_calls(text) tool_calls = self._extract_tool_calls(
text,
round_num=attempt,
on_debug=on_debug,
)
finalize_call: dict | None = None finalize_call: dict | None = None
other_calls: list[dict] = [] other_calls: list[dict] = []
@ -722,27 +835,67 @@ class GameEngine:
names = [tc.get("tool", "?") for tc in tool_calls] names = [tc.get("tool", "?") for tc in tool_calls]
round_log.append(f" tools: {', '.join(names)}") round_log.append(f" tools: {', '.join(names)}")
# Guard: no get tools alongside finalize_turn
get_tools = {"read_file", "character_get", "world_get", "journal_get"}
if finalize_call and any(tc.get("tool") in get_tools for tc in other_calls):
round_log.append(" mixed get + finalize — rejected")
debug_entries.append("\n".join(round_log))
messages = messages[:2]
messages.append({"role": "assistant", "content": text})
messages.append({
"role": "user",
"content": "## Validation Error\nYou used a get tool (`read_file`, `character_get`, `world_get`, `journal_get`) and `finalize_turn` in the same round. Decide: either gather information (use get tools, then stop), or finalize the turn (call `finalize_turn` alone with all data). Do not mix them."
})
if on_debug:
on_debug("validation_error", {"round": attempt, "type": "mixed_get_finalize", "tools": [tc.get("tool") for tc in other_calls]})
round_used += 1
continue
# finalize_turn present → validate and return # finalize_turn present → validate and return
if finalize_call: if finalize_call:
args = finalize_call.get("args", {}) args = finalize_call.get("args", {})
errs = [] errs = []
if not args.get("book_log"): if not args.get("book_log"):
errs.append("book_log is required") errs.append("book_log [Required]")
if not args.get("user_prompt"): if not args.get("user_prompt"):
errs.append("user_prompt is required") errs.append("user_prompt [Required]")
if errs: if errs:
hint = (
f"Expected:\n"
f'{{"tool": "finalize_turn", "args": {{'
f'"book_log": "...", '
f'"user_prompt": "...", '
f'"log_entry": "...", '
f'"ambience": "..."'
f"}}}}\n"
)
round_log.append(f" finalize_turn validation errors: {', '.join(errs)}") round_log.append(f" finalize_turn validation errors: {', '.join(errs)}")
debug_entries.append("\n".join(round_log)) debug_entries.append("\n".join(round_log))
messages = messages[:2]
messages.append({"role": "assistant", "content": text}) messages.append({"role": "assistant", "content": text})
messages.append({ messages.append({
"role": "user", "role": "user",
"content": f"## Validation Error\nfinalize_turn missing: {', '.join(errs)}. Please provide all required fields and call finalize_turn again." "content": f"## Validation Error\nMissing required field(s): {', '.join(errs)}.\n\n{hint}Please provide all required fields and call finalize_turn again."
}) })
if on_debug:
on_debug("validation_error", {"round": attempt, "type": "finalize_turn", "errors": errs})
round_used += 1
continue continue
if on_debug:
on_debug("finalize", {"round": attempt, "args": args})
round_used += 1
self._append_llm_log(
f"\n--- FINALIZE (attempt {attempt}) ---\n"
f"book_log: {args.get('book_log','')[:200]}\n"
f"user_prompt: {args.get('user_prompt','')[:200]}\n"
f"log_entry: {args.get('log_entry','')}\n"
f"ambience: {args.get('ambience','')}\n"
)
return TurnResult( return TurnResult(
book_log=args.get("book_log", ""), book_log=args.get("book_log", ""),
user_prompt=args.get("user_prompt", ""), user_prompt=args.get("user_prompt", ""),
ambience=args.get("ambience"), ambience=args.get("ambience"),
log_entry=args.get("log_entry"),
) )
# Execute other tools # Execute other tools
@ -756,15 +909,19 @@ class GameEngine:
if not args.get("dm_status"): if not args.get("dm_status"):
err_msg = ( err_msg = (
f"**Validation Error:** Tool `{name}` missing required `dm_status`. " f"**Validation Error:** Tool `{name}` missing required `dm_status`. "
f"Describe what the DM is doing (e.g. " f"Add `\"dm_status\": \"what the DM is doing\"` to the args.\n"
f'`"dm_status": "consulting the archives"`). Please retry.' f"Put each tool call in its own ```tool block."
) )
results.append(err_msg) results.append(err_msg)
round_log.append(f" {name}: MISSING dm_status") round_log.append(f" {name}: MISSING dm_status")
if on_debug:
on_debug("validation_error", {"round": attempt, "type": "tool", "tool": name, "error": "missing dm_status"})
continue continue
if on_action: if on_action:
on_action(self._describe_tool_action(name, args)) on_action(self._describe_tool_action(name, args))
if on_debug:
on_debug("tool_call", {"round": attempt, "tool": name, "args": args})
if name == "player_roll" and on_player_roll: if name == "player_roll" and on_player_roll:
dice = args.get("dice", "1d6") dice = args.get("dice", "1d6")
reason = args.get("reason", "a check") reason = args.get("reason", "a check")
@ -774,24 +931,55 @@ class GameEngine:
result = self._execute_tool(name, args) result = self._execute_tool(name, args)
results.append(f"**Tool:** {name}\n**Args:** {json.dumps(args)}\n**Result:** {result}") results.append(f"**Tool:** {name}\n**Args:** {json.dumps(args)}\n**Result:** {result}")
round_log.append(f" {name}: OK") round_log.append(f" {name}: OK")
if on_debug:
on_debug("tool_result", {"round": attempt, "tool": name, "result": result})
messages = messages[:2]
messages.append({"role": "assistant", "content": text}) messages.append({"role": "assistant", "content": text})
messages.append({ messages.append({
"role": "user", "role": "user",
"content": "## Tool Results\n\n" + "\n\n".join(results), "content": "## Tool Results\n\n" + "\n\n".join(results),
}) })
debug_entries.append("\n".join(round_log)) debug_entries.append("\n".join(round_log))
round_used += 1
continue continue
# No tools, no finalize → remind LLM # No tools, no finalize
round_log.append(" no tool calls — prompted to use tools") round_log.append(" no tool calls")
if not text.strip():
# Empty response — model may be slow. Give it time and retry without adding context.
if on_debug:
on_debug("empty_response", {"round": attempt})
import time
time.sleep(2)
debug_entries.append("\n".join(round_log)) debug_entries.append("\n".join(round_log))
continue
# Plain-text reasoning (no ```tool/```thought blocks) — log in debug but don't show to player
round_used += 1
if on_debug:
on_debug("thought", {"round": attempt, "text": text.strip()})
debug_entries.append("\n".join(round_log))
messages = messages[:2]
messages.append({"role": "assistant", "content": text}) messages.append({"role": "assistant", "content": text})
messages.append({ reminder_count += 1
"role": "user", if reminder_count % 3 == 0:
"content": "## Instructions\nUse tools to gather information or call **finalize_turn** to complete the turn." reminder = (
}) "## Instructions\n"
"Respond with tool calls or finalize_turn.\n\n"
"Put each tool call in its own ```tool block:\n"
"```tool\n{\"tool\": \"character_get\", \"args\": {\"dm_status\": \"...\"}}\n```\n\n"
"When ready, call **finalize_turn** with `book_log` and `user_prompt`."
)
else:
reminder = "Use tools to gather information or call **finalize_turn** to end the turn."
messages.append({"role": "user", "content": reminder})
if on_debug:
on_debug("no_tool_calls", {"round": attempt})
debug_text = "\n\n".join(debug_entries) debug_text = "\n\n".join(debug_entries)
self._append_llm_log(f"\n--- LOOP EXCEEDED ({max_rounds} rounds) ---\n{debug_text}")
return TurnResult( 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}", 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, debug_info=debug_text,
@ -892,6 +1080,12 @@ class GameEngine:
with open(log_path, "a") as f: with open(log_path, "a") as f:
f.write(entry.strip() + "\n") f.write(entry.strip() + "\n")
def _append_llm_log(self, text: str) -> None:
"""Append raw LLM activity to llm.log for debugging."""
LLM_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(LLM_LOG_PATH, "a") as f:
f.write(text + "\n")
def _update_journal( def _update_journal(
self, add: list[str] | None = None, done: list[str] | None = None self, add: list[str] | None = None, done: list[str] | None = None
) -> None: ) -> None:

View File

@ -7,6 +7,7 @@ Owns the TUI and game loop. Layout:
""" """
from __future__ import annotations from __future__ import annotations
import json
import os import os
import random import random
import sys import sys
@ -181,6 +182,7 @@ class AmbiencePlayer:
self._options = {} self._options = {}
self._device = None self._device = None
self._stream = None self._stream = None
self._muted = False
self.load_options() self.load_options()
@property @property
@ -191,6 +193,17 @@ class AmbiencePlayer:
def ambience_name(self): def ambience_name(self):
return self.current_ambience return self.current_ambience
@property
def is_muted(self):
return self._muted
def toggle_mute(self):
self._muted = not self._muted
if self._muted:
self._stop()
else:
self._load_current()
def load_options(self): def load_options(self):
self._options = parse_ambience_options() self._options = parse_ambience_options()
@ -217,16 +230,23 @@ class AmbiencePlayer:
name = AMBIENCE_PATH.read_text().strip().lower() name = AMBIENCE_PATH.read_text().strip().lower()
except OSError: except OSError:
return return
self._switch_to(name) # Save the name even when muted — will play on unmute
self.current_ambience = name
self._stop()
if not self._muted and name != 'silence' and name in self._options:
self._play_current()
def _switch_to(self, name): def _switch_to(self, name):
if name == self.current_ambience: if name == self.current_ambience:
return return
self.current_ambience = name self.current_ambience = name
self._stop() self._stop()
if name == 'silence' or name not in self._options: if self._muted or name == 'silence' or name not in self._options:
return return
tracks = self._options.get(name, []) self._play_current()
def _play_current(self):
tracks = self._options.get(self.current_ambience, [])
valid = [t for t in tracks if t.exists()] valid = [t for t in tracks if t.exists()]
if not valid: if not valid:
return return
@ -238,6 +258,11 @@ class AmbiencePlayer:
except Exception: except Exception:
self.current_ambience = None self.current_ambience = None
def _load_current(self):
"""Called on unmute — replay current ambience if not silence."""
if self.current_ambience and self.current_ambience != 'silence':
self._play_current()
# module-level ref # module-level ref
app_ambience_player = None app_ambience_player = None
@ -343,6 +368,31 @@ class TranscriptPane(AutoStatic):
self.parent.scroll_end(animate=False) self.parent.scroll_end(animate=False)
class DebugPane(Static):
"""Scrolling log of LLM thoughts, tool calls, and results for this turn."""
MAX_LINES = 200
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._lines: list[str] = []
def append(self, text: str) -> None:
self._lines.append(text)
if len(self._lines) > self.MAX_LINES:
self._lines.pop(0)
self.update("\n".join(self._lines[-100:]))
self.call_after_refresh(self._scroll_bottom)
def _scroll_bottom(self):
if self.parent and hasattr(self.parent, 'scroll_end'):
self.parent.scroll_end(animate=False)
def clear(self) -> None:
self._lines.clear()
self.update("")
class CharPane(AutoStatic): class CharPane(AutoStatic):
def load(self): def load(self):
if not CHAR_PATH.exists(): if not CHAR_PATH.exists():
@ -422,6 +472,20 @@ class ChaosTUI(App):
color: #c8c8c8; color: #c8c8c8;
padding: 0 1; padding: 0 1;
} }
#debug-content {
background: #1a1a1a;
color: #88b0a0;
padding: 0 1;
}
#debug-content .dm-thought {
color: #c0a060;
}
#debug-content .dm-tool {
color: #60a0c0;
}
#debug-content .dm-result {
color: #80a080;
}
/* Play tab */ /* Play tab */
#play-narrative { #play-narrative {
@ -515,6 +579,25 @@ class ChaosTUI(App):
height: 1; height: 1;
text-style: italic; text-style: italic;
} }
#mute-btn {
dock: bottom;
width: 6;
height: 1;
background: #2a2a2a;
color: #888888;
border: none;
padding: 0 1;
min-width: 6;
margin: 0;
}
#mute-btn:hover {
background: #3a3a3a;
color: #cccccc;
}
#mute-btn.muted {
color: #ff6b6b;
text-style: bold;
}
""" """
BINDINGS = [ BINDINGS = [
@ -553,6 +636,9 @@ class ChaosTUI(App):
self._book_pages = [] self._book_pages = []
self._prev_page_count = 0 self._prev_page_count = 0
# Debug log
self._debug_lines: list[str] = []
# ── Compose ────────────────────────────────────────── # ── Compose ──────────────────────────────────────────
def compose(self): def compose(self):
yield Static(f"⚔ The Chaos ╎ {TODAY}", id="banner") yield Static(f"⚔ The Chaos ╎ {TODAY}", id="banner")
@ -585,7 +671,11 @@ class ChaosTUI(App):
yield Button("Next >>", id="book-next") yield Button("Next >>", id="book-next")
with VerticalScroll(id="book-scroll"): with VerticalScroll(id="book-scroll"):
yield Static("", id="book-content") yield Static("", id="book-content")
with TabPane("DEBUG", id="debug-tab"):
with VerticalScroll():
yield DebugPane("", id="debug-content")
yield StatusBar(id="status-bar") yield StatusBar(id="status-bar")
yield Button("", id="mute-btn", classes="mute-button")
def on_mount(self): def on_mount(self):
ensure_log() ensure_log()
@ -593,6 +683,7 @@ class ChaosTUI(App):
self._init_book() self._init_book()
self.set_interval(REFRESH_SECS, self._check_ambience) self.set_interval(REFRESH_SECS, self._check_ambience)
self.set_interval(REFRESH_SECS, self._reload_book) self.set_interval(REFRESH_SECS, self._reload_book)
self.call_after_refresh(self._update_mute_button)
# Start the game # Start the game
self.call_after_refresh(self._begin_game) self.call_after_refresh(self._begin_game)
@ -617,6 +708,23 @@ class ChaosTUI(App):
if app_ambience_player: if app_ambience_player:
app_ambience_player.poll() app_ambience_player.poll()
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "mute-btn":
if app_ambience_player:
app_ambience_player.toggle_mute()
self._update_mute_button()
def _update_mute_button(self) -> None:
btn = self.query_one("#mute-btn", Button)
if app_ambience_player and app_ambience_player.is_muted:
btn.label = "♪ muted"
btn.classes = "muted"
btn.tooltip = "Unmute music"
else:
btn.label = ""
btn.classes = ""
btn.tooltip = "Mute music"
# ── Game Loop ───────────────────────────────────────── # ── Game Loop ─────────────────────────────────────────
def _call_llm(self, player_action: str | None = None): def _call_llm(self, player_action: str | None = None):
"""Called when the player has acted — sends their action to the LLM.""" """Called when the player has acted — sends their action to the LLM."""
@ -630,6 +738,14 @@ class ChaosTUI(App):
self._show_thinking() self._show_thinking()
# Clear debug for new turn
pane = self.query_one("#debug-content", DebugPane)
pane.clear()
if player_action:
self._append_debug(f"▶ player action: {player_action}")
else:
self._append_debug("▶ starting new turn")
# Run generation in a daemon thread so it doesn't block the UI # Run generation in a daemon thread so it doesn't block the UI
t = threading.Thread( t = threading.Thread(
target=self._run_generation, target=self._run_generation,
@ -648,16 +764,27 @@ class ChaosTUI(App):
def on_action(action: str) -> None: def on_action(action: str) -> None:
self.call_from_thread(self._on_action, action) self.call_from_thread(self._on_action, action)
def on_debug(event_type: str, data: dict) -> None:
self.call_from_thread(self._on_debug, event_type, data)
result = self.engine.generate_with_tools( result = self.engine.generate_with_tools(
player_action=player_action, player_action=player_action,
last_prompt=last_prompt, last_prompt=last_prompt,
on_thought=on_thought, on_thought=on_thought,
on_action=on_action, on_action=on_action,
on_player_roll=self._on_player_roll, on_player_roll=self._on_player_roll,
on_debug=on_debug,
) )
self.call_from_thread(self._on_generation_done, result, player_action) self.call_from_thread(self._on_generation_done, result, player_action)
def _append_debug(self, text: str) -> None:
"""Append a line to the debug pane."""
from datetime import datetime
ts = datetime.now().strftime("%H:%M:%S")
pane = self.query_one("#debug-content", DebugPane)
pane.append(f"[{ts}] {text}")
def _show_thinking(self) -> None: def _show_thinking(self) -> None:
"""Show the thinking indicator and start the animation timer.""" """Show the thinking indicator and start the animation timer."""
self._dm_action = "DM is weaving the narrative" self._dm_action = "DM is weaving the narrative"
@ -688,6 +815,7 @@ class ChaosTUI(App):
status.add_class("processing") status.add_class("processing")
spinner = self._spinner_frames[0] spinner = self._spinner_frames[0]
status.update(f"{spinner} {display}") status.update(f"{spinner} {display}")
self._append_debug(f"{display}")
def _on_action(self, action: str) -> None: def _on_action(self, action: str) -> None:
"""Display a DM action (tool call) in the status bar.""" """Display a DM action (tool call) in the status bar."""
@ -697,9 +825,57 @@ class ChaosTUI(App):
status.add_class("processing") status.add_class("processing")
spinner = self._spinner_frames[0] spinner = self._spinner_frames[0]
status.update(f"{spinner} {action}") status.update(f"{spinner} {action}")
self._append_debug(action)
def _on_debug(self, event_type: str, data: dict) -> None:
"""Structured debug entry: visible description + technical detail."""
r = data.get("round", "")
if event_type == "llm_response":
text = data.get("text", "")
if text.strip():
preview = text[:200].replace("\n", "\\n").strip() + ("" if len(text) > 200 else "")
self._append_debug(f" LLM response: {preview}")
else:
self._append_debug(f" LLM response: (empty)")
elif event_type == "thought":
thought = data.get("text", "")
display = thought[:60] + "" if len(thought) > 60 else thought
self._append_debug(f" 💭 {display}")
elif event_type == "tool_call":
tool = data.get("tool", "?")
args = data.get("args", {})
desc = args.get("dm_status", tool)
self._append_debug(f" 🔧 {desc}")
self._append_debug(f" {tool}({json.dumps(args)})")
elif event_type == "tool_result":
tool = data.get("tool", "?")
result = data.get("result", "")
preview = result[:80].replace("\n", " ").strip() + ("" if len(result) > 80 else "")
self._append_debug(f"{preview}")
elif event_type == "validation_error":
err_type = data.get("type", "")
if err_type == "finalize_turn":
self._append_debug(f" ✖ finalize_turn missing: {', '.join(data.get('errors', []))}")
elif err_type == "mixed_get_finalize":
tools = data.get("tools", [])
self._append_debug(f" ✖ mixed get tools {tools} with finalize_turn — rejected")
else:
tool = data.get("tool", "?")
self._append_debug(f"{tool} missing dm_status")
elif event_type == "finalize":
self._append_debug(" ✔ finalize_turn")
elif event_type == "no_tool_calls":
self._append_debug(f" ⚠ no tool calls — reminded to use tools")
elif event_type == "parse_error":
self._append_debug(f" ⚠ failed to parse tool block: {data.get('content', '')}")
elif event_type == "empty_response":
self._append_debug(" ⚠ empty response — waiting 2s, retrying without reminder")
elif event_type == "llm_error":
self._append_debug(f" ✖ LLM error: {data.get('error', '')}")
def _on_player_roll(self, dice: str, reason: str) -> str: def _on_player_roll(self, dice: str, reason: str) -> str:
"""Called from worker thread. Shows roll popup, blocks until player responds.""" """Called from worker thread. Shows roll popup, blocks until player responds."""
self.call_from_thread(self._append_debug, f"🎲 asks player to roll {dice} ({reason})")
self.call_from_thread(self._show_roll_modal, dice, reason) self.call_from_thread(self._show_roll_modal, dice, reason)
self._roll_event.wait() self._roll_event.wait()
self._roll_event.clear() self._roll_event.clear()
@ -736,8 +912,21 @@ class ChaosTUI(App):
if result.error: if result.error:
self._show_error(result.error, result.debug_info) self._show_error(result.error, result.debug_info)
self._append_debug(f"✖ error: {result.error}")
return return
# Log only after successful finalize — failed turns produce no side effects
from datetime import datetime
ts = datetime.now().strftime("%H:%M")
if player_action:
time_of_day = self._guess_time_of_day()
self.engine.append_log(f"- **{time_of_day}** — {player_action}")
if result.log_entry:
self.engine.append_log(f"- **{ts}** — {result.log_entry}")
elif result.book_log:
first_line = result.book_log.strip().split("\n")[0][:80]
self.engine.append_log(f"- **Turn** — {first_line}")
# Archive the turn's book log # Archive the turn's book log
if result.book_log: if result.book_log:
self.engine.archive_turn(result.book_log) self.engine.archive_turn(result.book_log)
@ -755,6 +944,7 @@ class ChaosTUI(App):
# Store for next turn # Store for next turn
self._last_prompt = result.user_prompt self._last_prompt = result.user_prompt
self._last_result = result self._last_result = result
self._append_debug("✔ turn complete")
def _display_scene(self, result: TurnResult) -> None: def _display_scene(self, result: TurnResult) -> None:
"""Update the UI with the last story entry followed by the DM prompt.""" """Update the UI with the last story entry followed by the DM prompt."""
@ -800,14 +990,6 @@ class ChaosTUI(App):
def _handle_player_action(self, action: str) -> None: def _handle_player_action(self, action: str) -> None:
"""Handle a player action typed in the input.""" """Handle a player action typed in the input."""
# Log the action
from datetime import datetime
timestamp = datetime.now().strftime("%H:%M")
time_of_day = self._guess_time_of_day()
log_entry = f"- **{time_of_day}** — {action}"
self.engine.append_log(log_entry)
# Call LLM to resolve
self._call_llm(player_action=action) self._call_llm(player_action=action)
def _guess_time_of_day(self) -> str: def _guess_time_of_day(self) -> str: