"""Chat panel widget for the TUI.""" from __future__ import annotations import json from typing import Any from rich.console import RenderableType from rich.text import Text from textual.app import ComposeResult from textual.containers import ScrollableContainer from textual.widgets import Static from clients.terminal.tui.chat_model import ChatItem, ChatModel from clients.terminal.tui.renderers import default_registry from clients.terminal.tui.settings import get_tui_settings from clients.terminal.tui.themes import get_active_theme def _item_msg(item: ChatItem) -> dict[str, Any]: """Map a ChatItem to the msg dict its renderer expects.""" kind = item.kind if kind == "user_message": return {"type": "user_message", "content": item.content} if kind == "assistant_message": return {"type": "assistant_message", "content": item.content} if kind == "thinking_block": return { "type": "thinking_block", "content": item.content, "is_subagent": item.meta.get("is_subagent", False), } if kind == "tool_started": return {"type": "tool_started", **item.meta} if kind == "tool_call": return {"type": "tool_call", **item.meta} if kind == "error": return {"type": "error", "message": item.content} if kind == "status": return {"type": "status", "content": item.content} if kind == "planning_status": return { "type": "planning_status", "label": item.content, "phase": item.meta.get("phase"), "is_subagent": item.meta.get("is_subagent", False), } if kind == "plan_ready": return { "type": "plan_ready", "plan": item.content, "is_subagent": item.meta.get("is_subagent", False), } if kind == "turn_meta": return {"type": "turn_meta", "elapsed_seconds": item.meta.get("elapsed_seconds")} return {"type": "plain", "content": item.content} def _signature(msg: dict[str, Any]) -> str: """Stable serialized form of a renderer msg, for cache invalidation. Only an item whose rendered inputs changed should re-render; this captures content + meta so in-place mutations (e.g. a streaming assistant bubble growing token by token, or a rolling planning_status line) invalidate exactly their own widget. """ return json.dumps(msg, sort_keys=True, default=str) class _ChatItemView(Static): """One chat item as its own widget. Making each message a distinct widget is what keeps streaming cheap: a per-token ``stream_delta`` only re-renders *this* widget (via ``Static.update``), not the whole conversation — Textual refreshes just the changed region instead of re-laying-out every visible item. """ DEFAULT_CSS = """ _ChatItemView { height: auto; width: 1fr; padding: 0; } """ def __init__(self, renderable: RenderableType, *, item: ChatItem, signature: str) -> None: super().__init__(renderable) self._item: ChatItem = item self._signature: str = signature def maybe_update(self, item: ChatItem, registry) -> bool: """Re-render only if the item's signature changed. Returns True if updated.""" msg = _item_msg(item) sig = _signature(msg) if sig == self._signature: return False self._item = item self._signature = sig self.update(registry.render(msg)) return True class ChatPanel(ScrollableContainer): """Scrollable conversation panel. Each visible message is a separate ``_ChatItemView`` child widget rather than one big ``Static`` rebuilt on every event. On a per-token stream the only widget that re-renders is the streaming bubble; everything else is untouched. Items are mounted/unmounted incrementally as the model grows and as the visible-window cap truncates the oldest. """ DEFAULT_CSS = """ ChatPanel { border: solid $tui-border; background: $tui-surface; color: $tui-text; padding: 0 1; height: 1fr; width: 2fr; } ChatPanel Static { color: $tui-text; } """ def __init__(self) -> None: super().__init__() self._model = ChatModel() self._registry = default_registry() # Truncation hint shown at the top when the history exceeds the cap. self._hint_view = Static("", id="chat-truncation-hint") self._hint_view.styles.display = "none" # Number of older items currently truncated out of view (0 = none). self._hidden_count = 0 # item -> view widget, keyed by id(item). Only holds visible items, so # it stays bounded by the window cap, not by total history length. self._widgets: dict[int, _ChatItemView] = {} def compose(self) -> ComposeResult: yield self._hint_view def add_user_message(self, text: str) -> None: self._model.add_user_message(text) self._sync() def load_history(self, messages: list[dict]) -> None: """Populate the chat panel with a session's stored messages on resume.""" self._model.load_history(messages) self._rebuild_all() def handle_ws_event(self, msg: dict) -> None: self._model.handle_ws_event(msg) self._sync() def clear(self) -> None: """Reset the chat model and redraw an empty conversation.""" self._model.items.clear() self._model._current_assistant = None self._model._current_thinking = None self._rebuild_all() def _max_visible_items(self) -> int: return get_tui_settings().max_visible_items def _truncation_hint(self, hidden: int) -> RenderableType: theme = get_active_theme() return Text( f"… {hidden} earlier messages not shown", style=theme.text.hex, ) def _rebuild_all(self) -> None: """Unmount every item widget and re-sync from scratch. Used on session switch / clear, where the model is replaced wholesale: the surviving-vs-new diff is meaningless when nothing survives. """ for widget in self._widgets.values(): widget.remove() self._widgets.clear() self._sync() def _sync(self) -> None: """Reconcile mounted widgets with the model's visible window. The model only ever appends new items and pops/purges existing ones — it never reorders survivors — so new widgets always belong at the end and a simple mount-at-end keeps the DOM in the right order. Surviving items keep their widget and are only re-rendered if their signature changed (in-place mutation, e.g. a growing stream). """ items = self._model.items limit = self._max_visible_items() hidden = max(0, len(items) - limit) # Slice the last `limit` items when over the cap; otherwise keep all. visible = items[-limit:] if hidden else items visible_ids = {id(it) for it in visible} # Truncation hint at the top of the window. self._hidden_count = hidden if hidden: self._hint_view.update(self._truncation_hint(hidden)) self._hint_view.styles.display = "block" else: self._hint_view.styles.display = "none" # Drop widgets whose item left the visible window (truncated out, or # purged from the model by stream_end / plan_ready). for key in [k for k in self._widgets if k not in visible_ids]: widget = self._widgets.pop(key) widget.remove() # Walk the visible window in order: mount new items (at the end) and # re-render changed ones in place. for item in visible: key = id(item) widget = self._widgets.get(key) if widget is None: msg = _item_msg(item) widget = _ChatItemView( self._registry.render(msg), item=item, signature=_signature(msg), ) self._widgets[key] = widget self.mount(widget) else: widget.maybe_update(item, self._registry) self.scroll_end(animate=False)