"""The main window and the worker thread a turn runs on. Turn logic is `console.Session` — the same object the terminal uses — driven on a QThread with a `Bridge` standing in for the terminal. The window only ever reacts to Bridge signals, so nothing here knows about HTTP or tools. """ from __future__ import annotations import os import sys import threading from typing import Optional from PySide6.QtCore import QObject, Qt, QThread, QTimer, Signal, Slot from PySide6.QtGui import QFont, QKeySequence, QShortcut from PySide6.QtWidgets import ( QApplication, QFileDialog, QFrame, QHBoxLayout, QLabel, QMainWindow, QScrollArea, QSizePolicy, QVBoxLayout, QWidget, ) from .. import __version__, config, dictation, server_client from ..console import Session from ..tools import Tools from . import icons, theme from .bridge import Bridge from .chat_list_window import ChatListDialog, ChatSignals, run_in_pool from .widgets import ( AssistantMessage, Composer, EmptyState, FolderChip, HoldingLine, ModeSwitch, Notice, StatusDot, ThinkingIndicator, ToolCard, UserMessage, ) class TurnRunner(QObject): """Lives on the worker thread; runs one Session.send per request.""" finished = Signal() def __init__(self, session: Session): super().__init__() self.session = session @Slot(str) def run(self, text: str) -> None: try: self.session.send(text) except Exception as exc: # never let a bug kill the worker thread self.session.ui.error(f"Unexpected error: {exc}") self.session.ui.stop_spinner() finally: self.finished.emit() class HealthProbe(QObject): done = Signal(bool, str) def start(self) -> None: threading.Thread(target=self._probe, daemon=True).start() def _probe(self) -> None: try: server_client.check_health(timeout=8) self.done.emit(True, f"Connected to {config.SERVER_URL}") except Exception as exc: self.done.emit(False, f"Can't reach {config.SERVER_URL}: {exc}") class DictationSignals(QObject): """Dictation calls back on background threads; these queue it onto the GUI thread.""" update = Signal(str, str) # final text so far, current interim guess level = Signal(float) # 0..1 mic loudness done = Signal(str, str) # text, error ("" when none) stopping = Signal() # recording over, transcript on its way class Column(QWidget): """Centres its content in a reading column of at most COLUMN_WIDTH.""" def __init__(self, parent=None): super().__init__(parent) row = QHBoxLayout(self) row.setContentsMargins(24, 0, 24, 0) row.addStretch(1) self.inner = QWidget() self.inner.setMaximumWidth(theme.COLUMN_WIDTH) self.inner.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred) row.addWidget(self.inner, 100) row.addStretch(1) self.layout_ = QVBoxLayout(self.inner) self.layout_.setContentsMargins(0, 0, 0, 0) class MainWindow(QMainWindow): submit_turn = Signal(str) HINT = "Enter to send · Shift+Enter for a new line · Ctrl+M dictate · Ctrl+N new chat" def __init__(self, mode: Optional[str] = None, session: Optional[Session] = None, dictation_factory=None, list_chats=None, create_chat=None): super().__init__() # Injectable so tests never hit the network; the real ones run on the pool. self._list_chats = list_chats or server_client.list_chats self._create_chat = create_chat or server_client.create_chat self._dictation_factory = dictation_factory or dictation.Dictation self._dictation = None self.setWindowTitle("Bolt") self.setWindowIcon(icons.app_icon()) self.resize(1040, 800) self.setMinimumSize(620, 480) self.bridge = Bridge() if session is None: session = Session(self.bridge, tools=Tools(self.bridge, mode=mode or config.APPROVAL_MODE)) self.session = session self._cards = {} self._assistant: Optional[AssistantMessage] = None self._busy = False self._stick_to_bottom = True root = QWidget() root.setObjectName("Root") self.setCentralWidget(root) page = QVBoxLayout(root) page.setContentsMargins(0, 0, 0, 0) page.setSpacing(0) page.addWidget(self._build_header()) # transcript self.scroll = QScrollArea() self.scroll.setWidgetResizable(True) self.scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) transcript = QWidget() transcript.setObjectName("Transcript") body = QVBoxLayout(transcript) body.setContentsMargins(0, 18, 0, 18) self.column = Column() body.addWidget(self.column) body.addStretch(1) self.entries = self.column.layout_ self.entries.setSpacing(6) self.empty = EmptyState() self.empty.picked.connect(self._send) self.entries.addWidget(self.empty) self.thinking = ThinkingIndicator() self.thinking.hide() self.entries.addWidget(self.thinking) self.scroll.setWidget(transcript) self.scroll.verticalScrollBar().rangeChanged.connect(self._follow) self.scroll.verticalScrollBar().valueChanged.connect(self._track_scroll) page.addWidget(self.scroll, 1) # composer dock = Column() dock.layout_.setContentsMargins(0, 6, 0, 14) dock.layout_.setSpacing(6) self.composer = Composer() self.composer.submitted.connect(self._send) dock.layout_.addWidget(self.composer) self.hint = QLabel(self.HINT) self.hint.setObjectName("Hint") self.hint.setAlignment(Qt.AlignHCenter) dock.layout_.addWidget(self.hint) self._hint_timer = QTimer(self) self._hint_timer.setSingleShot(True) self._hint_timer.timeout.connect(lambda: self.hint.setText(self.HINT)) page.addWidget(dock) QShortcut(QKeySequence("Ctrl+L"), self, activated=self.clear_transcript) QShortcut(QKeySequence("Ctrl+O"), self, activated=self.choose_folder) QShortcut(QKeySequence("Ctrl+N"), self, activated=self.start_new_chat) QShortcut(QKeySequence("Ctrl+K"), self, activated=self.open_chats) self.chat_signals = ChatSignals() self.chat_signals.created.connect(lambda session_id: self.switch_chat(session_id, "New chat")) self.chat_signals.failed.connect(self._new_chat_failed) self._creating_chat = False QShortcut(QKeySequence("Ctrl+M"), self, activated=self.toggle_dictation) QShortcut(QKeySequence("Escape"), self, activated=self.cancel_dictation) self.dictation_signals = DictationSignals() self.dictation_signals.update.connect(self.composer.show_dictation) self.dictation_signals.level.connect(self.composer.mic.set_level) self.dictation_signals.done.connect(self._dictation_done) self.dictation_signals.stopping.connect( lambda: self._dictation is not None and self.composer.mic.set_state("finishing")) self.composer.mic.clicked.connect(self.toggle_dictation) self._mic_unavailable = dictation.unavailable_reason() if self._mic_unavailable: self.composer.mic.set_state("unavailable", self._mic_unavailable) self._wire_bridge() self._start_worker() self._refresh_folder() self.health = HealthProbe() self.health.done.connect(self._health_done) QTimer.singleShot(0, self.health.start) self.composer.edit.setFocus() # ── construction ───────────────────────────────────────────────────── def _build_header(self) -> QWidget: header = QFrame() header.setObjectName("Header") header.setFixedHeight(56) row = QHBoxLayout(header) row.setContentsMargins(18, 0, 18, 0) row.setSpacing(10) row.addWidget(icons.IconLabel("bolt", theme.ACCENT_INK, 26, tile=theme.ACCENT)) word = QLabel("Bolt") word.setObjectName("Wordmark") row.addWidget(word) sub = QLabel(f"Console {__version__}") sub.setObjectName("WordmarkSub") row.addWidget(sub) row.addSpacing(6) self.status = StatusDot() self.status.set_state("unknown", "Checking the server…") row.addWidget(self.status) row.addStretch(1) self.chats = FolderChip(icon_name="chats", text="Chats") self.chats.setToolTip("All your chats with Bolt, from every device (Ctrl+K)") self.chats.clicked.connect(self.open_chats) row.addWidget(self.chats) self.new_chat = FolderChip(icon_name="plus", text="New chat") self.new_chat.setToolTip("Start a fresh conversation (Ctrl+N)\n" "Fresh history; Bolt still remembers you") self.new_chat.clicked.connect(self.start_new_chat) row.addWidget(self.new_chat) self.folder = FolderChip() self.folder.clicked.connect(self.choose_folder) row.addWidget(self.folder) self.mode = ModeSwitch(self.session.tools.mode) self.mode.changed.connect(self._set_mode) row.addWidget(self.mode) return header def _wire_bridge(self) -> None: b = self.bridge b.busy.connect(self._set_busy) b.chunk.connect(self._on_chunk) b.reply_ended.connect(self._end_assistant) b.reply.connect(self._on_reply) b.holding_line.connect(lambda text: self._add(HoldingLine(text))) b.tool_started.connect(self._on_tool_started) b.tool_output.connect(self._on_tool_output) b.tool_diff.connect(self._on_tool_diff) b.approval_needed.connect(self._on_approval) b.notice.connect(lambda level, text: self._add(Notice(level, text))) def _start_worker(self) -> None: self.thread_ = QThread(self) self.runner = TurnRunner(self.session) self.runner.moveToThread(self.thread_) self.submit_turn.connect(self.runner.run) self.runner.finished.connect(self._turn_finished) self.thread_.start() # ── transcript helpers ─────────────────────────────────────────────── def _add(self, widget: QWidget) -> None: if self.empty.isVisible(): self.empty.hide() # Always insert above the thinking indicator, which stays last. self.entries.insertWidget(self.entries.count() - 1, widget) def _track_scroll(self, value: int) -> None: bar = self.scroll.verticalScrollBar() self._stick_to_bottom = value >= bar.maximum() - 40 def _follow(self, _min: int, maximum: int) -> None: if self._stick_to_bottom: self.scroll.verticalScrollBar().setValue(maximum) def clear_transcript(self) -> None: if self._busy: return keep = (self.empty, self.thinking) for i in reversed(range(self.entries.count())): widget = self.entries.itemAt(i).widget() if widget is not None and widget not in keep: self.entries.takeAt(i) widget.deleteLater() self._cards.clear() self._assistant = None self.empty.show() # ── user actions ───────────────────────────────────────────────────── # ── chats ──────────────────────────────────────────────────────────── def start_new_chat(self) -> None: """Ask the server for a fresh chat id, off the UI thread; the switch happens when it arrives (switch_chat).""" if self._busy or self._creating_chat: return self._creating_chat = True self.new_chat.setEnabled(False) self._flash_hint("Starting a new chat...") run_in_pool("create", self._create_chat, self.chat_signals) def _new_chat_failed(self, _operation: str, message: str) -> None: self._creating_chat = False self.new_chat.setEnabled(not self._busy) self._flash_hint(f"Couldn't start a new chat: {message}") def open_chats(self) -> None: if self._busy: self._flash_hint("Wait for Bolt to finish before switching chats.") return dialog = ChatListDialog(self, list_chats=self._list_chats, create_chat=self._create_chat) dialog.chosen.connect(self.switch_chat) self.chat_dialog = dialog # kept for tests; the dialog is modal dialog.open() def switch_chat(self, session_id: str, label: str = "") -> None: """Point every server call at *session_id* and start the local transcript over: there's no endpoint for a past chat's messages, so the window can't show them (Bolt still has them server-side).""" self._creating_chat = False self.new_chat.setEnabled(not self._busy) if self._busy: return note = self.session.switch_to(session_id) self.clear_transcript() self.chats.setToolTip(f"All your chats with Bolt, from every device (Ctrl+K)\n{note}") self._flash_hint(f"Now in: {label or session_id}") def _send(self, text: str) -> None: text = text.strip() if not text or self._busy: return self._stick_to_bottom = True self._add(UserMessage(text)) self._assistant = None self._set_busy(True) self.submit_turn.emit(text) # ── dictation ──────────────────────────────────────────────────────── def toggle_dictation(self) -> None: if self._mic_unavailable: self._flash_hint(self._mic_unavailable) return if self._dictation is not None: if self.composer.mic.state == "recording": self.composer.mic.set_state("finishing") self._dictation.stop() return sig = self.dictation_signals recording = self._dictation_factory( sig.update.emit, sig.level.emit, lambda text, error: sig.done.emit(text, error or ""), silence_seconds=config.DICTATION_SILENCE_SECONDS, on_stopping=sig.stopping.emit, ) self.composer.begin_dictation() try: recording.start() except Exception as exc: self.composer.end_dictation("") self._flash_hint(f"Couldn't open the microphone: {exc}") return self._dictation = recording self.composer.mic.set_state("recording") if not recording.live: self._flash_hint("Live transcription is unavailable; the words will appear when you stop.") def cancel_dictation(self) -> None: if self._dictation is not None: self._dictation.cancel() def _dictation_done(self, text: str, error: str) -> None: self._dictation = None self.composer.mic.set_state("unavailable" if self._mic_unavailable else "idle", self._mic_unavailable or "") self.composer.end_dictation(text) if error: self._flash_hint(error) def _flash_hint(self, text: str, seconds: float = 6.0) -> None: self.hint.setText(text) self._hint_timer.start(int(seconds * 1000)) def choose_folder(self) -> None: if self._busy: return path = QFileDialog.getExistingDirectory(self, "Working directory for Bolt", os.getcwd()) if path: try: os.chdir(path) except OSError as exc: self._add(Notice("error", f"Couldn't switch to {path}: {exc}")) return self._refresh_folder() if not self.empty.isVisible(): self._add(Notice("info", f"Working directory is now {os.getcwd()}")) def _refresh_folder(self) -> None: cwd = os.getcwd() self.folder.set_path(cwd) self.empty.set_folder(cwd) def _set_mode(self, mode: str) -> None: self.session.tools.mode = mode # ── bridge slots ───────────────────────────────────────────────────── def _set_busy(self, busy: bool) -> None: self._busy = busy self.composer.set_busy(busy) self.folder.setEnabled(not busy) self.new_chat.setEnabled(not busy and not self._creating_chat) self.chats.setEnabled(not busy) if busy: self.thinking.start() else: self.thinking.stop() def _turn_finished(self) -> None: self._set_busy(False) self._assistant = None # Pick up a mode switched by "Always allow" on the worker side. self.mode.set_mode(self.session.tools.mode) self.composer.edit.setFocus() def _on_chunk(self, sentence: str) -> None: if self._assistant is None: self._assistant = AssistantMessage() self._add(self._assistant) self._assistant.append(sentence) def _end_assistant(self) -> None: self._assistant = None def _on_reply(self, text: str) -> None: self._add(AssistantMessage(text)) self._assistant = None def _on_tool_started(self, tool_id: int, name: str, argument: str) -> None: self._assistant = None # text after a tool call starts a new block card = ToolCard(name, argument) card.answered.connect(self._on_answer) self._cards[tool_id] = card self._add(card) def _on_tool_output(self, tool_id: int, text: str, error: bool) -> None: card = self._cards.get(tool_id) if card is not None: card.set_result(text, error) def _on_tool_diff(self, tool_id: int, old: str, new: str) -> None: card = self._cards.get(tool_id) if card is not None: card.set_diff(old, new) def _on_approval(self, tool_id: int, question: str) -> None: card = self._cards.get(tool_id) if card is None: self.bridge.answer("no") return self.thinking.stop() card.ask(question) self._stick_to_bottom = True QTimer.singleShot(0, lambda: self.scroll.ensureWidgetVisible(card, 0, 40)) QApplication.alert(self) # flash the taskbar if the window is behind others def _on_answer(self, value: str) -> None: if value == "always": self.mode.set_mode("auto") self.thinking.start() self.bridge.answer(value) def _health_done(self, ok: bool, detail: str) -> None: self.status.set_state("ok" if ok else "error", detail) # ── window chrome ──────────────────────────────────────────────────── def resizeEvent(self, event) -> None: super().resizeEvent(event) # The transcript grows from the top; give the empty state the whole # viewport so its stretches can centre it. self.empty.setMinimumHeight(max(0, self.scroll.viewport().height() - 40)) def showEvent(self, event) -> None: super().showEvent(event) theme.dark_title_bar(self) def closeEvent(self, event) -> None: self.cancel_dictation() self.bridge.close() self.thread_.quit() self.thread_.wait(1500) super().closeEvent(event) def make_app(argv=None) -> QApplication: app = QApplication.instance() or QApplication(argv or sys.argv) app.setApplicationName("Bolt Console") app.setWindowIcon(icons.app_icon()) app.setStyle("Fusion") # a neutral base, so the stylesheet decides the look app.setPalette(theme.palette()) font = QFont(theme.ui_font_family()) font.setPointSizeF(10.5) font.setHintingPreference(QFont.PreferNoHinting) app.setFont(font) app.setStyleSheet(theme.stylesheet()) return app def run(argv=None) -> int: if sys.platform == "win32": try: # own taskbar group + icon instead of python.exe's import ctypes ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("TheMajesticNetwork.BoltConsole") except Exception: pass argv = list(sys.argv[1:] if argv is None else argv) app = make_app() if not config.is_configured(): from PySide6.QtWidgets import QMessageBox QMessageBox.critical( None, "Bolt Console", "Missing " + ", ".join(config.missing_config()) + ".\n\nCopy .env.example to .env next to the app and fill them in.") return 2 window = MainWindow(mode="auto" if "--auto" in argv else None) window.show() return app.exec()