#!/usr/bin/env python3 """ run.py — The Chaos TTRPG Session Client (Game Mode) Owns the TUI and game loop. Layout: PLAY (narrative + choices + input) | CHAR | LOG | BOOK tabs """ from __future__ import annotations import json import threading from textual import on from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical, VerticalScroll from textual.widgets import Button, Input, Static, TabbedContent, TabPane from rich.markdown import Markdown as RichMarkdown from rich.theme import Theme from engine import GameEngine from engine_lib.models import TurnResult from engine_lib import state from run_utils import ( BOOK_PATH, CHAR_PATH, CHANGES_PATH, SETTINGS_PATH, TODAY, REFRESH_SECS, clear_llm_log, ensure_log, load_book_pages, ) from run_ambience import AmbiencePlayer from run_widgets import ( app_ambience_player as _widget_player_ref, RollModal, DebugPane, CharPane, StatusBar, TodoPane, TranscriptPane, ) # ── Global state ───────────────────────────────────────── app_ambience_player: AmbiencePlayer | None = None # ── UI Theme ───────────────────────────────────────────── MARKDOWN_THEME = Theme({ "markdown.h1": "bold #ff6b6b on #2a0000", "markdown.h2": "bold #ffd93d", "markdown.h3": "bold italic #6bcbff", "markdown.h4": "bold #95e1d3", "markdown.code": "bold #00d2d3 on #002222", "markdown.code_block": "on #1e1e2e", "markdown.block_quote": "dim italic #8395a7", "markdown.link": "underline #48dbfb", "markdown.item": "#ff9ff3", "markdown.strong": "bold #feca57", "markdown.horizontal_rule": "dim #555555", }) # ── The App ────────────────────────────────────────────── class ChaosTUI(App): TITLE = "The Chaos" CSS = """ Screen { background: #121212; } #banner { dock: top; height: 1; background: #2a2a2a; color: #e0ad4c; text-align: center; } #main { height: 100%; background: #111111; } #todo-header { background: #3a2d23; color: #e0b060; padding: 0 1; height: 1; } #todo-content { background: #1a1510; color: #d0b080; padding: 0 1; height: 5; max-height:5; overflow-y:auto; scrollbar-size-vertical:2; } #main-tabs { height: 1fr; } TabbedContent { background: #1a1a2a; } VerticalScroll { overflow-y: auto; scrollbar-size-vertical:2; scrollbar-color:#555; scrollbar-color-hover:#777; scrollbar-color-active:#999; } #char-content { background: #1e1e2a; color: #c0c0c0; padding: 0 1; } #transcript { background: #1a2a1a; color: #c8c8c8; 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-narrative { background: #161616; color: #d8d8d8; padding: 1 2; height: auto; } #play-status { background: #1a2a1a; color: #e0b060; padding: 0 2; height: 1; text-style: bold italic; text-align: center; } #play-status.processing { background: #2a1a0a; color: #ffd93d; } #play-input { height: 3; background: #222; color: #e0d0c0; border: solid #555; padding: 0 1; } #play-input:focus { border: solid #e0ad4c; } #play-input:disabled { background: #1a1a1a; color: #666; border: solid #333; } #book-header { background: #2d2d2d; color: #e0c080; text-style: bold; padding: 0 1; height: 1; } #book-nav { height: 3; background: #222; align: center middle; } #book-nav Button { width: 10; margin: 0 1; } #book-nav Button:disabled { color: #444; } #book-nav Button:hover { text-style: bold; } #book-nav-center { height: 3; width: 1fr; } #book-page-label { height: 1; color: #c0b090; text-style: bold; padding: 0 2; text-align: center; } #book-progress { height: 1; background: #1a1a1a; color: #e0b060; padding: 0 2; text-align: center; } #book-scroll { height: 1fr; } #book-content { background: #161616; color: #d8d8d8; padding: 0 2; } #status-bar { background: #222; color: #888; padding: 0 1; height: 1; text-style: italic; } #mute-btn { dock: bottom; width: 6; height: 1; background: #2a2a2a; color: #888; border: none; padding: 0 1; min-width: 6; margin: 0; } #mute-btn:hover { background: #3a3a3a; color: #ccc; } #mute-btn.muted { color: #ff6b6b; text-style: bold; } """ BINDINGS = [ ("ctrl+c", "quit", "Quit"), ("escape", "quit", "Quit"), ] def __init__(self, *args, no_music=False, **kwargs): super().__init__(*args, **kwargs) global app_ambience_player app_ambience_player = None if no_music else AmbiencePlayer() import run_widgets run_widgets.app_ambience_player = app_ambience_player self.engine = GameEngine() self._last_result: TurnResult | None = None self._is_processing: bool = False self._spinner_frames = ["◴", "◷", "◶", "◵"] self._thinking_frame = 0 self._thinking_timer_handle = None self._dm_action = "DM is preparing a response" self._roll_event = threading.Event() self._roll_result: str | None = None self._book_page = 0 self._book_pages: list[str] = [] self._prev_page_count = 0 self._settings_loaded = False def _load_settings(self) -> dict: defaults = {"active_tab": "play-tab", "music_muted": False, "book_page": 0} if SETTINGS_PATH.exists(): try: data = json.loads(SETTINGS_PATH.read_text()) if isinstance(data, dict): defaults.update(data) except (json.JSONDecodeError, OSError): pass return defaults def _save_settings(self) -> None: if not self._settings_loaded: return active = self.query_one("#main-tabs", TabbedContent).active or "play-tab" data = { "active_tab": active, "music_muted": app_ambience_player.is_muted if app_ambience_player else False, "book_page": self._book_page, } try: SETTINGS_PATH.write_text(json.dumps(data, indent=2) + "\n") except OSError: pass def compose(self): yield Static(f"⚔ The Chaos ╎ {TODAY}", id="banner") with Vertical(id="main"): yield Static("TODO", id="todo-header") yield TodoPane(id="todo-content") with TabbedContent(initial="play-tab", id="main-tabs"): with TabPane("PLAY", id="play-tab"): with VerticalScroll(id="play-scroll"): yield Static("*Awaiting the fates...*", id="play-narrative") yield Static("", id="play-status") yield Input(placeholder="Type your action and press Enter...", id="play-input") with TabPane("CHARACTER", id="char-tab"): with VerticalScroll(): yield CharPane(id="char-content") with TabPane("LOG", id="log-tab"): with VerticalScroll(): yield TranscriptPane(id="transcript") with TabPane("BOOK", id="book-tab"): yield Static("BOOK", id="book-header") with Vertical(id="book-section"): with Horizontal(id="book-nav"): yield Button("<< Prev", id="book-prev") with Vertical(id="book-nav-center"): yield Static("", id="book-page-label") yield Static("", id="book-progress") yield Button("Next >>", id="book-next") with VerticalScroll(id="book-scroll"): yield Static("", id="book-content") with TabPane("DEBUG", id="debug-tab"): with VerticalScroll(): yield DebugPane("", id="debug-content") yield StatusBar(id="status-bar") yield Button("♫", id="mute-btn", classes="mute-button") def on_mount(self): clear_llm_log() ensure_log() self.console._theme = MARKDOWN_THEME self._init_book() self.set_interval(REFRESH_SECS, self._check_ambience) self.set_interval(REFRESH_SECS, self._reload_book) self.call_after_refresh(self._update_mute_button) self.call_after_refresh(self._apply_settings) self.call_after_refresh(self._begin_game) def _apply_settings(self) -> None: settings = self._load_settings() self._book_page = settings.get("book_page", 0) self._prev_page_count = len(self._book_pages) self._reload_book() if settings.get("music_muted") and app_ambience_player and not app_ambience_player.is_muted: app_ambience_player.toggle_mute() self._update_mute_button() try: self.query_one("#main-tabs", TabbedContent).active = settings.get("active_tab", "play-tab") except Exception: pass self._settings_loaded = True def _begin_game(self): self._last_narrative: str = "" pages = load_book_pages() if pages and pages != ["*The story has not begun.*"]: parts = [] parts.append(pages[-1]) if CHANGES_PATH.exists(): saved = [l.strip() for l in CHANGES_PATH.read_text().splitlines() if l.strip()] if saved: parts.append(self._render_changes(saved)) self._set_narrative("\n\n".join(parts)) self._enable_input() return self._call_llm() def _check_ambience(self): if app_ambience_player: 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() self._save_settings() 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" def _call_llm(self, player_action: str | None = None): if self._is_processing: return self._is_processing = True input_widget = self.query_one("#play-input", Input) input_widget.disabled = True input_widget.placeholder = "DM is at work..." self._show_thinking() self.query_one("#debug-content", DebugPane).clear() self._append_debug(f"▶ {'player action: ' + player_action if player_action else 'starting new turn'}") t = threading.Thread(target=self._run_generation, args=(player_action,), daemon=True) t.start() def _run_generation(self, player_action: str | None) -> None: import traceback def on_thought(thought: str) -> None: self.call_from_thread(self._on_thought, thought) def on_action(action: str) -> None: 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) try: result = self.engine.generate_turn( player_action=player_action, on_thought=on_thought, on_action=on_action, on_player_roll=self._on_player_roll, on_debug=on_debug, ) except Exception as e: self.call_from_thread(self._on_generation_error, e, traceback.format_exc()) return self.call_from_thread(self._on_generation_done, result, player_action) def _append_debug(self, text: str) -> None: from datetime import datetime ts = datetime.now().strftime("%H:%M:%S") self.query_one("#debug-content", DebugPane).append(f"[{ts}] {text}") def _show_thinking(self) -> None: self._dm_action = "DM is preparing a response" self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") status.update(f"✦ {self._spinner_frames[0]} {self._dm_action} ✦") self._thinking_timer_handle = self.set_interval(0.5, self._tick_thinking) def _hide_thinking(self) -> None: if self._thinking_timer_handle: self._thinking_timer_handle.stop() self._thinking_timer_handle = None status = self.query_one("#play-status", Static) status.remove_class("processing") status.update("") def _on_thought(self, thought: str) -> None: display = thought[:60] + "…" if len(thought) > 60 else thought self._dm_action = display self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") status.update(f"✦ {self._spinner_frames[0]} {display} ✦") self._append_debug(f"✦ {display}") def _on_action(self, action: str) -> None: self._dm_action = action self._thinking_frame = 0 status = self.query_one("#play-status", Static) status.add_class("processing") status.update(f"✦ {self._spinner_frames[0]} {action} ✦") self._append_debug(action) def _on_debug(self, event_type: str, data: dict) -> None: if event_type == "phase": p = data.get("phase", 0) status = data.get("status", "") if status == "start": n = data.get("name", "") d = f" dice={data['dice']}" if data.get("dice") else "" o = f" [attempt {data['outer_attempt']}/3]" if data.get("outer_attempt") else "" self._append_debug(f"▸ Phase {p}: {n}{o}{d}") elif status == "done": if p == 1: self._append_debug(f" ✔ prose: {data.get('chars', 0)} chars") elif p == 2: self._append_debug(f" ✔ summary: {data.get('summary', '')}") elif p == 3: self._append_debug(f" ✔ extract: {data.get('applied', 0)} state changes applied") elif status == "empty": self._append_debug(f" ⚠ phase {p} attempt {data.get('attempt', '?')} empty — retry") elif status == "fallback": self._append_debug(f" ⚠ phase {p} used fallback: {data.get('summary', '')}") elif status == "tools_found": tools = data.get("tools", []) t = ", ".join(tools) if tools else "none" self._append_debug(f" 🔧 tools: {t}" + (" + finalize" if data.get("has_finalize") else "")) elif status == "errors": for e in data.get("errors", []): self._append_debug(f" ✖ {e}") self._append_debug(f" ⟳ retry ({data.get('attempt', '?')})") elif status == "exhausted": self._append_debug(" ✖ Phase 3 exhausted retries — state changes may be missing!") for e in data.get("errors", []): self._append_debug(f" {e}") elif status == "retry_after_phase3_failure": self._append_debug(f" ⟳ Phase 3 failed — retry from Phase 1 (attempt {data.get('outer_attempt', '?')}/3)") elif status == "validation_failed": self._append_debug(f" ✖ narrative rejected: {data.get('reason', '?')} (attempt {data.get('outer_attempt', '?')}/3)") elif event_type == "phase_done": self._append_debug(f" ✔ turn — book_log: {data.get('book_log_chars', 0)} chars") if data.get("log_entry"): self._append_debug(f" log: {data['log_entry']}") if data.get("ambience"): self._append_debug(f" ambience: {data['ambience']}") if data.get("extract_errors"): self._append_debug(f" extract errors: {data['extract_errors']}") elif event_type == "config": m = data.get("model", "?") t = data.get("temperature", "?") tk = data.get("max_tokens", "?") s = data.get("strategy", "?") self._append_debug(f"▸ LLM: {m} | temp={t} | tokens={tk} | strategy={s}") elif event_type == "tool_call": self._append_debug(f" 🔧 {data['tool']}({json.dumps(data.get('args', {}))})") elif event_type == "tool_result": r = data.get("result", "") self._append_debug(f" → {r[:80].replace(chr(10),' ').strip()}{'…' if len(r)>80 else ''}") elif event_type == "parse_error": self._append_debug(f" ⚠ bad tool block: {data.get('content', '')}") elif event_type == "llm_error": self._append_debug(f" ✖ LLM [{data.get('label','')}]: {data.get('error','')}") elif event_type == "turn_details": ts = data.get("timestamp", "") m = data.get("model", "?") t = data.get("temperature", "?") tk = data.get("max_tokens", "?") s = data.get("strategy_name", "?") d = data.get("die_roll", "?") ic = len(data.get("player_action", "")) oc = data.get("book_log_chars", 0) w = data.get("book_log_words", 0) a = data.get("ambience", "None") tc = data.get("tool_calls_count", 0) ap = data.get("applied_changes_count", 0) tms = data.get("total_elapsed_ms", 0) ams = data.get("apply_elapsed_ms", 0) self._append_debug(f" ━━━ Turn ━━━ {ts}") self._append_debug(f" LLM: {m} | temp={t} | tokens={tk} | strategy={s}") self._append_debug(f" Dice: {d} | In: {ic}c | Out: {oc}c ({w}w)") self._append_debug(f" Amb: {a} | Tools: {tc} (applied: {ap}) | Time: {tms:.0f}ms ({ams:.0f}ms apply)") for tc2 in data.get("tool_call_results", []): self._append_debug(f" 🔧 {tc2['tool']}: {json.dumps(tc2.get('args',{}))[:120]}") def _on_player_roll(self, dice: str, reason: str) -> str: self.call_from_thread(self._append_debug, f"🎲 roll {dice} ({reason})") self.call_from_thread(self._show_roll_modal, dice, reason) self._roll_event.wait() self._roll_event.clear() result = self._roll_result self._roll_result = None return result or "0" def _show_roll_modal(self, dice: str, reason: str) -> None: self._roll_event.clear() self._roll_result = None def on_dismiss(value: str) -> None: self._roll_result = value self._roll_event.set() self.push_screen(RollModal(dice, reason), on_dismiss) def _tick_thinking(self) -> None: if not self._is_processing: return self._thinking_frame = (self._thinking_frame + 1) % len(self._spinner_frames) self.query_one("#play-status", Static).update( f"✦ {self._spinner_frames[self._thinking_frame]} {self._dm_action} ✦") def _on_generation_done(self, result: TurnResult, player_action: str | None) -> None: self._is_processing = False self._hide_thinking() if result.error: self._show_error(result.error, result.debug_info) self._append_debug(f"✖ error: {result.error}") return if result.book_log: turn_num = state.archive_turn(result.book_log) if result.log_entry: state.append_log(f"- **Turn {turn_num}** — {result.log_entry}") else: summary = result.book_log.strip().split(chr(10))[0][:80] state.append_log(f"- **Turn {turn_num}** — {summary}") elif result.log_entry: state.append_log(f"- {result.log_entry}") state.apply_state(result) if result.book_log or not result.user_prompt: self._display_scene(result) else: if self._last_narrative: self._set_narrative(f"{self._last_narrative}\n\n---\n\n{result.user_prompt}") self._enable_input() else: self._display_scene(result) if result.book_log: self._last_result = result self._append_debug("✔ turn complete") def _on_generation_error(self, error: Exception, traceback_str: str) -> None: self._is_processing = False self._hide_thinking() err_msg = f"{type(error).__name__}: {error}" self._append_debug(f"✖ UNHANDLED: {err_msg}") for line in traceback_str.rstrip().split("\n")[-10:]: self._append_debug(f" {line}") self._show_error(err_msg, traceback_str) @staticmethod def _render_changes(changes: list[str]) -> str: return "**Changes:**\n" + "\n".join(f"- {c}" for c in changes) def _display_scene(self, result: TurnResult) -> None: parts = [] if result.book_log: parts.append(result.book_log) if result.changes: parts.append(self._render_changes(result.changes)) if result.user_prompt: parts.append(f"---\n\n{result.user_prompt}") self._set_narrative("\n\n".join(parts) if parts else "") self._enable_input() def _enable_input(self, value: str = "") -> None: inp = self.query_one("#play-input", Input) inp.disabled = False inp.placeholder = "Type your action and press Enter..." inp.value = value inp.focus() def _set_narrative(self, text: str) -> None: self._last_narrative = text self.query_one("#play-narrative", Static).update(RichMarkdown(text)) self.query_one("#play-scroll", VerticalScroll).scroll_home(animate=False) def _show_error(self, error: str, debug_info: str = "") -> None: t = f"**Error:** {error}\n\n" + (f"**Debug Info:**\n\n{debug_info}\n\n" if debug_info else "") self._set_narrative(t + "Check your session/config.json and ensure your LLM provider is running.") self._enable_input(value=self._last_player_action if hasattr(self, '_last_player_action') else "") def on_input_submitted(self, event: Input.Submitted) -> None: action = event.value.strip() if not action or self._is_processing: event.stop() return event.stop() self._last_player_action = action self._call_llm(player_action=action) def _init_book(self): self._reload_book() self._render_book_page() def _reload_book(self): self._book_pages = load_book_pages() if len(self._book_pages) > self._prev_page_count: self._book_page = len(self._book_pages) - 1 self._prev_page_count = len(self._book_pages) self._book_page = max(0, min(self._book_page, len(self._book_pages) - 1)) self._render_book_page() def _render_book_page(self): if not self._book_pages: return self.query_one("#book-content").update(RichMarkdown(self._book_pages[self._book_page])) total = len(self._book_pages) self.query_one("#book-page-label").update(f"Page {self._book_page + 1} of {total}") pct = (self._book_page + 1) / total if total else 1 self.query_one("#book-progress").update(f"[{'█'*round(pct*20)}{'░'*(20-round(pct*20))}]") self.query_one("#book-prev").disabled = (self._book_page == 0) self.query_one("#book-next").disabled = (self._book_page == total - 1) def action_prev_page(self): if self._book_page > 0: self._book_page -= 1 self._render_book_page() self.query_one("#book-scroll").scroll_home(animate=False) def action_next_page(self): if self._book_page < len(self._book_pages) - 1: self._book_page += 1 self._render_book_page() self.query_one("#book-scroll").scroll_home(animate=False) @on(Button.Pressed, "#book-prev") def on_book_prev(self): self.action_prev_page() self._save_settings() @on(Button.Pressed, "#book-next") def on_book_next(self): self.action_next_page() self._save_settings() def on_tabbed_content_tab_activated(self, event: TabbedContent.TabActivated) -> None: self._save_settings() def main(): import argparse parser = argparse.ArgumentParser(description="The Chaos TUI") parser.add_argument('--no-music', action='store_true', help='Disable ambience music') ChaosTUI(no_music=parser.parse_args().no_music).run() if __name__ == '__main__': main()