diff options
Diffstat (limited to 'llamachat/ui.py')
| -rw-r--r-- | llamachat/ui.py | 902 |
1 files changed, 902 insertions, 0 deletions
diff --git a/llamachat/ui.py b/llamachat/ui.py new file mode 100644 index 0000000..d8ec10e --- /dev/null +++ b/llamachat/ui.py @@ -0,0 +1,902 @@ +# SPDX-License-Identifier: GPL-2.0-only +# +# llamachat - a small native chat client for a local llama.cpp router +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +"""The chat window.""" + +import html +import re +from pathlib import Path + +from PySide6.QtCore import QObject, Qt, QThread, Signal, Slot +from PySide6.QtGui import ( + QAction, QDesktopServices, QKeySequence, QTextDocument, +) +from PySide6.QtWidgets import ( + QButtonGroup, QComboBox, QFileDialog, QHBoxLayout, QLabel, QLineEdit, + QListWidget, QListWidgetItem, QMainWindow, QMenu, QMessageBox, + QPlainTextEdit, QPushButton, QRadioButton, QSplitter, QTextBrowser, + QVBoxLayout, QWidget, +) + +from . import backend +from .backend import Attachment, BackendError + +MODE_ONESHOT = "oneshot" +MODE_CHAT = "chat" + +# URL scheme for the reasoning toggle. A reply is untrusted text, so links +# it produces are stripped rather than trusted to not collide with this. +REASONING_SCHEME = "x-llamachat-reasoning:" + + +class StreamWorker(QObject): + """Runs one streaming request off the GUI thread.""" + + chunk = Signal(str) + reasoning = Signal(str) + finished = Signal() + failed = Signal(str) + + def __init__(self, client, model: str, messages: list[dict]): + super().__init__() + self.client = client + self.model = model + self.messages = messages + self._stop = False + + def stop(self) -> None: + self._stop = True + + @Slot() + def run(self) -> None: + try: + for kind, piece in self.client.stream_chat(self.model, self.messages): + if self._stop: + break + if kind == "reasoning": + self.reasoning.emit(piece) + else: + self.chunk.emit(piece) + except BackendError as exc: + self.failed.emit(str(exc)) + return + except Exception as exc: # noqa: BLE001 - surface anything as UI text + self.failed.emit(f"Unexpected error: {exc}") + return + self.finished.emit() + + +class ChatWindow(QMainWindow): + """Model picker, mode toggle, transcript, input, history panel.""" + + def __init__(self, cfg, history, client, presets): + super().__init__() + self.cfg = cfg + self.history = history + self.client = client + self.presets = presets + + self.session_id: int | None = None + self.mode = MODE_ONESHOT + self.read_only = False + self.attachments: list[Attachment] = [] + self.thread: QThread | None = None + self.worker: StreamWorker | None = None + self.assistant_message_id: int | None = None + self.assistant_buffer = "" + self.reasoning_buffer = "" + # Every rendered bubble, so a reasoning toggle can redraw the + # transcript without refetching anything. + self.bubbles: list[dict] = [] + self.expanded: set[int] = set() + + self.setWindowTitle("llamachat") + self.resize(1000, 700) + self.setAcceptDrops(True) + self._build_ui() + self.refresh_models() + self.refresh_history() + + # -- construction ----------------------------------------------------- + + def _build_ui(self) -> None: + central = QWidget() + outer = QVBoxLayout(central) + + # Top bar: model, mode, search. + top = QHBoxLayout() + self.model_box = QComboBox() + self.model_box.setMinimumWidth(220) + self.model_box.currentTextChanged.connect(self._on_model_changed) + top.addWidget(QLabel("Model:")) + top.addWidget(self.model_box) + + self.reload_button = QPushButton("⟳") + self.reload_button.setToolTip("Reload model list from the router") + self.reload_button.setFixedWidth(32) + self.reload_button.clicked.connect(self.refresh_models) + top.addWidget(self.reload_button) + + top.addSpacing(16) + self.oneshot_radio = QRadioButton("One-shot") + self.chat_radio = QRadioButton("Chat") + self.oneshot_radio.setChecked(True) + group = QButtonGroup(self) + group.addButton(self.oneshot_radio) + group.addButton(self.chat_radio) + self.oneshot_radio.toggled.connect(self._on_mode_changed) + top.addWidget(self.oneshot_radio) + top.addWidget(self.chat_radio) + + top.addStretch() + self.search_box = QLineEdit() + self.search_box.setPlaceholderText("Search history…") + self.search_box.setClearButtonEnabled(True) + self.search_box.setMaximumWidth(280) + self.search_box.textChanged.connect(self._on_search) + top.addWidget(self.search_box) + outer.addLayout(top) + + # Body: history list beside the transcript and input. + splitter = QSplitter(Qt.Horizontal) + + self.history_list = QListWidget() + self.history_list.setMinimumWidth(200) + self.history_list.itemActivated.connect(self._on_history_activated) + self.history_list.itemClicked.connect(self._on_history_activated) + self.history_list.setContextMenuPolicy(Qt.CustomContextMenu) + self.history_list.customContextMenuRequested.connect( + self._on_history_menu + ) + splitter.addWidget(self.history_list) + + right = QWidget() + right_layout = QVBoxLayout(right) + right_layout.setContentsMargins(0, 0, 0, 0) + + self.transcript = QTextBrowser() + # Links are handled here, not followed: reasoning:N toggles a + # thinking block, anything else opens in the browser. + self.transcript.setOpenLinks(False) + self.transcript.anchorClicked.connect(self._on_anchor_clicked) + right_layout.addWidget(self.transcript, 1) + + self.attach_label = QLabel() + self.attach_label.setWordWrap(True) + self.attach_label.hide() + right_layout.addWidget(self.attach_label) + + self.status_label = QLabel() + self.status_label.hide() + right_layout.addWidget(self.status_label) + + self.input = QPlainTextEdit() + self.input.setPlaceholderText( + "Type a message. Ctrl+Enter sends, Esc hides the window." + ) + self.input.setMaximumHeight(140) + right_layout.addWidget(self.input) + + buttons = QHBoxLayout() + self.new_button = QPushButton("New") + self.new_button.clicked.connect(self.new_session) + buttons.addWidget(self.new_button) + + self.attach_button = QPushButton("Attach…") + self.attach_button.clicked.connect(self._on_attach_clicked) + buttons.addWidget(self.attach_button) + + self.clear_attach_button = QPushButton("Clear files") + self.clear_attach_button.clicked.connect(self.clear_attachments) + self.clear_attach_button.hide() + buttons.addWidget(self.clear_attach_button) + + buttons.addStretch() + self.stop_button = QPushButton("Stop") + self.stop_button.clicked.connect(self.stop_stream) + self.stop_button.hide() + buttons.addWidget(self.stop_button) + + self.send_button = QPushButton("Send") + self.send_button.setDefault(True) + self.send_button.clicked.connect(self.send) + buttons.addWidget(self.send_button) + right_layout.addLayout(buttons) + + splitter.addWidget(right) + splitter.setStretchFactor(0, 0) + splitter.setStretchFactor(1, 1) + splitter.setSizes([240, 760]) + outer.addWidget(splitter, 1) + + self.setCentralWidget(central) + + send_action = QAction(self) + send_action.setShortcut(QKeySequence("Ctrl+Return")) + send_action.triggered.connect(self.send) + self.addAction(send_action) + + send_enter = QAction(self) + send_enter.setShortcut(QKeySequence("Ctrl+Enter")) + send_enter.triggered.connect(self.send) + self.addAction(send_enter) + + # -- model handling --------------------------------------------------- + + def refresh_models(self) -> None: + """Repopulate the picker from the router, keeping the selection.""" + previous = self.model_box.currentText() + try: + available = self.client.models() + except BackendError as exc: + self.show_status(str(exc), error=True) + return + + self.model_box.blockSignals(True) + self.model_box.clear() + for name in available: + preset = self.presets.get(name) + label = f"{name} 👁" if preset and preset.vision else name + self.model_box.addItem(label, name) + self.model_box.blockSignals(False) + + target = previous or self.cfg.default_model + if target: + index = self.model_box.findData(_strip_marker(target)) + if index < 0: + index = self.model_box.findText(target) + if index >= 0: + self.model_box.setCurrentIndex(index) + if available: + self.hide_status() + + def current_model(self) -> str: + return self.model_box.currentData() or self.model_box.currentText() + + def current_preset(self): + return self.presets.get(self.current_model()) + + def char_budget(self) -> int: + preset = self.current_preset() + if preset is None: + return int(4096 * self.cfg.chars_per_token * self.cfg.attach_ctx_fraction) + return preset.char_budget( + self.cfg.attach_ctx_fraction, self.cfg.chars_per_token + ) + + def vision_models(self) -> list[str]: + names = [] + for i in range(self.model_box.count()): + name = self.model_box.itemData(i) + preset = self.presets.get(name) + if preset and preset.vision: + names.append(name) + return names + + def _on_model_changed(self, _text: str) -> None: + if self.session_id is not None and not self.read_only: + self.history.touch_session(self.session_id, self.current_model()) + + # -- mode handling ---------------------------------------------------- + + def _on_mode_changed(self, _checked: bool) -> None: + new_mode = MODE_ONESHOT if self.oneshot_radio.isChecked() else MODE_CHAT + if new_mode == self.mode: + return + self.mode = new_mode + self.new_session() + + def new_session(self) -> None: + """Drop the current conversation and start clean.""" + self.session_id = None + self.read_only = False + self.assistant_message_id = None + self.assistant_buffer = "" + self.reasoning_buffer = "" + self.bubbles.clear() + self.expanded.clear() + self.clear_attachments() + self.transcript.clear() + self.input.clear() + self.input.setReadOnly(False) + self.send_button.setEnabled(True) + self.history_list.clearSelection() + self.hide_status() + + # -- attachments ------------------------------------------------------ + + def _on_attach_clicked(self) -> None: + paths, _ = QFileDialog.getOpenFileNames(self, "Attach files") + if paths: + self.add_attachments([Path(p) for p in paths]) + + def add_attachments(self, paths: list[Path]) -> None: + """Load files, refusing images unless the model can see them.""" + wants_vision = any(backend.classify(p) == "image" for p in paths) + if wants_vision and not self._ensure_vision_model(): + return + + for path in paths: + try: + att = backend.load_attachment(path, self.char_budget()) + except (BackendError, OSError) as exc: + self.show_status(f"{path.name}: {exc}", error=True) + continue + self.attachments.append(att) + if att.truncated: + self.show_status( + f"{path.name} was truncated to fit the model context " + f"({self.char_budget()} characters).", + error=True, + ) + self.update_attach_label() + + def _ensure_vision_model(self) -> bool: + """Offer to switch to a vision preset. True when one is active.""" + preset = self.current_preset() + if preset and preset.vision: + return True + + candidates = self.vision_models() + if not candidates: + QMessageBox.warning( + self, + "No vision model", + "No model on the router has an mmproj file configured, so " + "images cannot be sent.", + ) + return False + + target = candidates[0] + answer = QMessageBox.question( + self, + "Switch model?", + f"Images need a vision-capable model.\n\n" + f"Switch to {target}?\n\n" + "The router unloads the current model to do this, so the next " + "reply will take several seconds to start.", + QMessageBox.Yes | QMessageBox.No, + QMessageBox.Yes, + ) + if answer != QMessageBox.Yes: + return False + + index = self.model_box.findData(target) + if index >= 0: + self.model_box.setCurrentIndex(index) + return True + + def clear_attachments(self) -> None: + self.attachments.clear() + self.update_attach_label() + + def update_attach_label(self) -> None: + if not self.attachments: + self.attach_label.hide() + self.clear_attach_button.hide() + return + names = [] + for att in self.attachments: + mark = "🖼" if att.kind == "image" else "📄" + suffix = " (truncated)" if att.truncated else "" + names.append(f"{mark} {att.path.name}{suffix}") + self.attach_label.setText("Attached: " + ", ".join(names)) + self.attach_label.show() + self.clear_attach_button.show() + + # -- drag and drop ---------------------------------------------------- + + def dragEnterEvent(self, event) -> None: + if event.mimeData().hasUrls(): + event.acceptProposedAction() + + def dragMoveEvent(self, event) -> None: + if event.mimeData().hasUrls(): + event.acceptProposedAction() + + def dropEvent(self, event) -> None: + paths = [ + Path(url.toLocalFile()) + for url in event.mimeData().urls() + if url.isLocalFile() + ] + if paths: + self.add_attachments(paths) + event.acceptProposedAction() + + # -- sending ---------------------------------------------------------- + + def send(self) -> None: + if self.thread is not None: + return + if self.read_only: + self.show_status( + "One-shot entries are read-only. Press New to start again.", + error=True, + ) + return + + text = self.input.toPlainText().strip() + if not text and not self.attachments: + return + model = self.current_model() + if not model: + self.show_status("No model selected.", error=True) + return + + if self.session_id is None: + title = (text or self.attachments[0].path.name)[:60] + self.session_id = self.history.create_session( + self.mode, model, title + ) + + content = backend.build_user_content(text, self.attachments) + stored = content if isinstance(content, str) else content[0]["text"] + message_id = self.history.add_message(self.session_id, "user", stored) + for att in self.attachments: + self.history.add_attachment( + message_id, + path=str(att.path), + kind=att.kind, + mime=att.mime, + size=att.size, + sha256=att.sha256, + thumb=att.thumb, + truncated=att.truncated, + ) + + self.append_bubble("user", stored, self.attachments) + self.input.clear() + + if self.mode == MODE_CHAT: + messages = self._chat_context(content) + else: + messages = [{"role": "user", "content": content}] + + self.clear_attachments() + self.refresh_history() + self._start_stream(model, messages) + + def _chat_context(self, latest_content) -> list[dict]: + """Prior turns plus the new message, for multi-turn mode.""" + messages = [] + rows = self.history.messages(self.session_id) + for row in rows[:-1]: # the latest user row is replaced below + messages.append({"role": row["role"], "content": row["content"]}) + messages.append({"role": "user", "content": latest_content}) + return messages + + def _start_stream(self, model: str, messages: list[dict]) -> None: + self.assistant_buffer = "" + self.reasoning_buffer = "" + self.assistant_message_id = self.history.add_message( + self.session_id, "assistant", "" + ) + self.append_bubble("assistant", "") + self.show_status(f"Waiting for {model}… (a model switch takes a while)") + self.send_button.setEnabled(False) + self.stop_button.show() + + self.thread = QThread(self) + self.worker = StreamWorker(self.client, model, messages) + self.worker.moveToThread(self.thread) + self.thread.started.connect(self.worker.run) + self.worker.chunk.connect(self._on_chunk) + self.worker.reasoning.connect(self._on_reasoning) + self.worker.finished.connect(self._on_stream_finished) + self.worker.failed.connect(self._on_stream_failed) + self.thread.start() + + @Slot(str) + def _on_chunk(self, piece: str) -> None: + if not self.assistant_buffer: + self.hide_status() + self.assistant_buffer += piece + self._update_last_bubble(text=self.assistant_buffer) + + @Slot(str) + def _on_reasoning(self, piece: str) -> None: + if not self.reasoning_buffer: + self.hide_status() + self.reasoning_buffer += piece + self._update_last_bubble(reasoning=self.reasoning_buffer) + + @Slot() + def _on_stream_finished(self) -> None: + if self.assistant_message_id is not None: + self.history.update_message( + self.assistant_message_id, + self.assistant_buffer, + self.reasoning_buffer, + ) + self._teardown_stream() + self.refresh_history() + + @Slot(str) + def _on_stream_failed(self, message: str) -> None: + if self.assistant_message_id is not None: + self.history.update_message( + self.assistant_message_id, + self.assistant_buffer, + self.reasoning_buffer, + ) + self.show_status(message, error=True) + self._teardown_stream() + + def stop_stream(self) -> None: + if self.worker is not None: + self.worker.stop() + + def _teardown_stream(self) -> None: + if self.thread is not None: + self.thread.quit() + self.thread.wait(3000) + self.thread = None + self.worker = None + self.assistant_message_id = None + self.send_button.setEnabled(True) + self.stop_button.hide() + + # -- transcript rendering --------------------------------------------- + + def append_bubble( + self, + role: str, + text: str, + attachments: list[Attachment] | None = None, + saved_rows=None, + reasoning: str = "", + ) -> None: + self.bubbles.append( + { + "role": role, + "text": text, + "attachments": attachments, + "saved_rows": saved_rows, + "reasoning": reasoning, + } + ) + self._render() + + def _update_last_bubble( + self, text: str | None = None, reasoning: str | None = None + ) -> None: + """Rewrite the streaming assistant bubble.""" + if not self.bubbles: + return + if text is not None: + self.bubbles[-1]["text"] = text + if reasoning is not None: + self.bubbles[-1]["reasoning"] = reasoning + self._render(live=True) + + def _render(self, live: bool = False) -> None: + """Redraw the whole transcript. + + Rewriting everything keeps the collapse state and the streaming + update on one code path. Conversations are short enough that the + cost does not show. + """ + bar = self.transcript.verticalScrollBar() + at_end = bar.value() >= bar.maximum() - 4 + previous = bar.value() + + parts = [] + for i, bubble in enumerate(self.bubbles): + streaming = live and i == len(self.bubbles) - 1 + parts.append( + _bubble_html( + bubble["role"], + bubble["text"], + bubble["attachments"], + bubble["saved_rows"], + reasoning=bubble["reasoning"], + index=i, + expanded=i in self.expanded, + live=streaming and not bubble["text"], + ) + ) + self.transcript.setHtml("".join(parts)) + + if at_end: + self._scroll_to_end() + else: + bar.setValue(previous) + + def _on_anchor_clicked(self, url) -> None: + """Expand or collapse a thinking block.""" + target = url.toString() + if target.startswith("blocked:"): + return # a link a reply tried to forge + if not target.startswith(REASONING_SCHEME): + if url.scheme() in ("http", "https", "mailto"): + QDesktopServices.openUrl(url) + return + try: + index = int(target[len(REASONING_SCHEME) :]) + except ValueError: + return + if index in self.expanded: + self.expanded.discard(index) + else: + self.expanded.add(index) + self._render(live=self.thread is not None) + + def _scroll_to_end(self) -> None: + bar = self.transcript.verticalScrollBar() + bar.setValue(bar.maximum()) + + # -- history panel ---------------------------------------------------- + + def refresh_history(self) -> None: + if self.search_box.text().strip(): + return # search results are showing; leave them alone + self.history_list.clear() + for row in self.history.recent_sessions(): + marker = "💬" if row["mode"] == MODE_CHAT else "⚡" + item = QListWidgetItem(f"{marker} {row['title'] or '(untitled)'}") + item.setData(Qt.UserRole, row["id"]) + item.setToolTip(f"{row['model'] or '?'} — {row['mode']}") + self.history_list.addItem(item) + + def _on_search(self, text: str) -> None: + if not text.strip(): + self.refresh_history() + return + self.history_list.clear() + for row in self.history.search(text): + item = QListWidgetItem( + f"{row['title'] or '(untitled)'}\n {_plain(row['snip'])}" + ) + item.setData(Qt.UserRole, row["session_id"]) + self.history_list.addItem(item) + + def _on_history_activated(self, item: QListWidgetItem) -> None: + session_id = item.data(Qt.UserRole) + if session_id is not None: + self.open_session(int(session_id)) + + def _on_history_menu(self, point) -> None: + item = self.history_list.itemAt(point) + if item is None: + return + menu = QMenu(self) + delete = menu.addAction("Delete") + if menu.exec(self.history_list.mapToGlobal(point)) == delete: + self.history.delete_session(int(item.data(Qt.UserRole))) + if self.session_id == int(item.data(Qt.UserRole)): + self.new_session() + self.refresh_history() + + def open_session(self, session_id: int) -> None: + """Load a past conversation. Chat mode continues, one-shot is read-only.""" + session = self.history.get_session(session_id) + if session is None: + return + + self.session_id = session_id + self.mode = session["mode"] + self.read_only = self.mode == MODE_ONESHOT + self.clear_attachments() + + self.oneshot_radio.blockSignals(True) + self.chat_radio.blockSignals(True) + self.oneshot_radio.setChecked(self.mode == MODE_ONESHOT) + self.chat_radio.setChecked(self.mode == MODE_CHAT) + self.oneshot_radio.blockSignals(False) + self.chat_radio.blockSignals(False) + + if session["model"]: + index = self.model_box.findData(session["model"]) + if index >= 0: + self.model_box.setCurrentIndex(index) + + self.bubbles.clear() + self.expanded.clear() + for row in self.history.messages(session_id): + self.bubbles.append( + { + "role": row["role"], + "text": row["content"], + "attachments": None, + "saved_rows": self.history.attachments(row["id"]), + "reasoning": _column(row, "reasoning"), + } + ) + self._render() + self._scroll_to_end() + + self.input.setReadOnly(self.read_only) + self.send_button.setEnabled(not self.read_only) + if self.read_only: + self.show_status( + "One-shot entry, read-only. Press New to start a fresh chat." + ) + else: + self.hide_status() + + # -- status line ------------------------------------------------------ + + def show_status(self, text: str, error: bool = False) -> None: + self.status_label.setText(text) + self.status_label.setStyleSheet( + "color: palette(bright-text); background: palette(highlight); " + "padding: 4px; border-radius: 3px;" + if error + else "padding: 4px;" + ) + self.status_label.show() + + def hide_status(self) -> None: + self.status_label.hide() + + # -- window behaviour ------------------------------------------------- + + def keyPressEvent(self, event) -> None: + if event.key() == Qt.Key_Escape: + self.hide() + return + super().keyPressEvent(event) + + def closeEvent(self, event) -> None: + """Closing hides; the daemon keeps running for the next toggle.""" + event.ignore() + self.hide() + + +# GitHub dialect for tables and strikethrough; NoHTML so markup in a reply +# is shown literally instead of being interpreted. +MARKDOWN_FLAGS = ( + QTextDocument.MarkdownDialectGitHub | QTextDocument.MarkdownNoHTML +) + +# Qt renders fenced code as one <pre> per line with no background of its +# own. Giving those a tint is what makes a code block read as a block. +_PRE_STYLE = ( + "background:rgba(127,127,127,0.18);padding:1px 6px;margin:0;" + "font-family:monospace" +) + + +def _markdown_to_fragment(text: str) -> str: + """Render markdown to an HTML fragment safe to splice into the page. + + QTextDocument does the parsing, so there is no markdown dependency. Its + output is a whole document, and only the body content is wanted here. + """ + if not text: + return "" + doc = QTextDocument() + doc.setMarkdown(text, MARKDOWN_FLAGS) + html_text = doc.toHtml() + + start = html_text.find("<body") + if start == -1: + return html.escape(text).replace("\n", "<br>") + start = html_text.find(">", start) + end = html_text.rfind("</body>") + if start == -1 or end == -1: + return html.escape(text).replace("\n", "<br>") + fragment = html_text[start + 1 : end].strip() + + # A reply could contain [x](x-llamachat-reasoning:0), which markdown + # turns into a real anchor. Defuse those so only the toggles this code + # emits can drive the UI. + fragment = fragment.replace(f'href="{REASONING_SCHEME}', 'href="blocked:') + + # Tint code blocks, which Qt leaves unstyled. Qt emits one <pre> per + # line, so the tint has to land on every one of them to read as a block. + return _PRE_RE.sub(_tint_pre, fragment) + + +_PRE_RE = re.compile(r"<pre(\s+style=\"([^\"]*)\")?\s*>") + + +def _tint_pre(match: re.Match) -> str: + existing = (match.group(2) or "").strip() + if existing and not existing.endswith(";"): + existing += ";" + return f'<pre style="{existing}{_PRE_STYLE}">' + + +def _column(row, name: str, default: str = "") -> str: + """Read a column that may predate a schema migration.""" + try: + value = row[name] + except (IndexError, KeyError): + return default + return value if value is not None else default + + +def _strip_marker(name: str) -> str: + return name.replace(" 👁", "").strip() + + +def _plain(text: str) -> str: + """Flatten an FTS snippet into one line of plain text.""" + return ( + text.replace("<b>", "").replace("</b>", "").replace("\n", " ")[:80] + ) + + +def _reasoning_html(text: str, index: int, expanded: bool, live: bool) -> str: + """The thinking block: a clickable summary line, body only when open. + + `index` identifies which bubble the toggle link belongs to, so several + replies in one transcript expand independently. + """ + if not text: + return "" + lines = text.count("\n") + 1 + arrow = "▾" if expanded else "▸" + label = "thinking…" if live else f"thinking ({len(text)} chars, {lines} lines)" + header = ( + f'<a href="{REASONING_SCHEME}{index}" style="color:palette(mid);' + f'text-decoration:none">{arrow} {label}</a>' + ) + if not expanded: + return f'<div style="margin:2px 0">{header}</div>' + body = html.escape(text).replace("\n", "<br>") + return ( + f'<div style="margin:2px 0">{header}</div>' + f'<div style="margin:2px 0 6px 14px;color:palette(mid);' + f'font-style:italic">{body}</div>' + ) + + +def _bubble_html( + role: str, + text: str, + attachments: list[Attachment] | None = None, + saved_rows=None, + reasoning: str = "", + index: int = 0, + expanded: bool = False, + live: bool = False, +) -> str: + """One transcript entry as HTML.""" + label = {"user": "You", "assistant": "Model"}.get(role, role) + colour = "palette(highlight)" if role == "user" else "palette(mid)" + + # Replies are markdown; what the user typed is shown as typed, so an + # attached file's contents cannot be reflowed into headings and lists. + if role == "assistant": + body = _markdown_to_fragment(text) + else: + body = html.escape(text).replace("\n", "<br>") + + files = "" + if attachments: + names = ", ".join(html.escape(a.path.name) for a in attachments) + files = f"<br><i>files: {names}</i>" + elif saved_rows: + parts = [] + for row in saved_rows: + missing = "" if Path(row["path"]).exists() else " (missing)" + parts.append(html.escape(row["path"]) + missing) + files = f"<br><i>files: {', '.join(parts)}</i>" + + think = _reasoning_html(reasoning, index, expanded, live) + + if role == "assistant": + # Markdown produces block elements, so the speaker label sits on its + # own line rather than trying to lead the first paragraph. + return ( + f'<div style="margin:6px 0">{think}' + f'<div style="color:{colour}"><b>{label}:</b></div>' + f"{body}{files}</div>" + ) + return ( + f'<div style="margin:6px 0">{think}' + f'<b style="color:{colour}">{label}:</b> {body}{files}</div>' + ) |