"""Main PySide6 window for the Javis desktop MVP.""" from __future__ import annotations from html import escape from PySide6.QtCore import Qt, QThread, QTimer, Signal from PySide6.QtGui import QAction, QCloseEvent, QKeySequence, QShortcut from PySide6.QtWidgets import ( QComboBox, QFormLayout, QHBoxLayout, QLabel, QLineEdit, QListWidget, QListWidgetItem, QMainWindow, QMenu, QMessageBox, QPushButton, QSplitter, QTextBrowser, QTextEdit, QVBoxLayout, QWidget, ) from javis.core.provider_router import ApprovalChoice from javis.providers.base import ChatMessage from javis.security.privacy import PrivacyDecision from javis.ui.chat_controller import DesktopController from javis.ui.models import GenerationState, StreamEvent, StreamEventKind from javis.ui.workers import ApprovalBridge, StreamWorker class MessageEdit(QTextEdit): send_requested = Signal() def keyPressEvent(self, event) -> None: if event.key() in {Qt.Key_Return, Qt.Key_Enter} and not ( event.modifiers() & Qt.ShiftModifier ): self.send_requested.emit() event.accept() return super().keyPressEvent(event) class MainWindow(QMainWindow): def __init__( self, controller: DesktopController, approval_bridge: ApprovalBridge, ) -> None: super().__init__() self.controller = controller self.approval_bridge = approval_bridge self._thread: QThread | None = None self._worker: StreamWorker | None = None self._messages: list[ChatMessage] = [] self._draft_user = "" self._draft_assistant = "" self._thinking_dots = 0 self._build_ui() self._connect_signals() self._apply_style() self._open_initial_session() self._animation = QTimer(self) self._animation.timeout.connect(self._animate_state) self._animation.start(450) def _build_ui(self) -> None: self.setWindowTitle("Javis Desktop") self.resize(1360, 820) splitter = QSplitter(Qt.Horizontal) splitter.addWidget(self._build_sidebar()) splitter.addWidget(self._build_chat()) splitter.addWidget(self._build_status_panel()) splitter.setSizes([280, 760, 300]) self.setCentralWidget(splitter) self.statusBar().showMessage("Bereit") QShortcut(QKeySequence("Ctrl+N"), self, activated=self._new_session) QShortcut(QKeySequence("Ctrl+K"), self, activated=self._focus_search) QShortcut(QKeySequence("Esc"), self, activated=self._stop_generation) def _build_sidebar(self) -> QWidget: panel = QWidget() layout = QVBoxLayout(panel) self.new_button = QPushButton("Neue Sitzung") self.search_edit = QLineEdit() self.search_edit.setPlaceholderText("Sitzungen suchen …") self.session_list = QListWidget() self.session_list.setContextMenuPolicy(Qt.CustomContextMenu) layout.addWidget(self.new_button) layout.addWidget(self.search_edit) layout.addWidget(self.session_list, 1) return panel def _build_chat(self) -> QWidget: panel = QWidget() layout = QVBoxLayout(panel) self.chat_view = QTextBrowser() self.chat_view.setOpenExternalLinks(False) self.message_edit = MessageEdit() self.message_edit.setPlaceholderText("Nachricht an Javis …") self.message_edit.setMaximumHeight(130) buttons = QHBoxLayout() buttons.addStretch(1) self.stop_button = QPushButton("Stoppen") self.stop_button.setEnabled(False) self.send_button = QPushButton("Senden") buttons.addWidget(self.stop_button) buttons.addWidget(self.send_button) layout.addWidget(self.chat_view, 1) layout.addWidget(self.message_edit) layout.addLayout(buttons) return panel def _build_status_panel(self) -> QWidget: panel = QWidget() layout = QVBoxLayout(panel) title = QLabel("SYSTEMSTATUS") title.setObjectName("panelTitle") layout.addWidget(title) self.state_label = QLabel("Bereit") self.state_label.setObjectName("stateLabel") layout.addWidget(self.state_label) self.mode_combo = QComboBox() self.mode_combo.addItem("Hybrid/Auto", "auto") self.mode_combo.addItem("Lokal", "local") form = QFormLayout() form.addRow("Modus", self.mode_combo) self.status_labels: dict[str, QLabel] = {} for key, label in ( ("provider", "Provider"), ("ollama", "Ollama erreichbar"), ("model", "Lokales Modell"), ("privacy", "Datenschutz"), ("cloud", "Cloudaufrufe heute"), ("fallback", "Lokales Fallback"), ("data", "Datenverzeichnis"), ): value = QLabel("—") value.setWordWrap(True) self.status_labels[key] = value form.addRow(label, value) layout.addLayout(form) layout.addStretch(1) return panel def _connect_signals(self) -> None: self.new_button.clicked.connect(self._new_session) self.search_edit.textChanged.connect(self._refresh_sessions) self.session_list.itemDoubleClicked.connect(self._load_item) self.session_list.customContextMenuRequested.connect(self._session_menu) self.send_button.clicked.connect(self._send) self.stop_button.clicked.connect(self._stop_generation) self.message_edit.send_requested.connect(self._send) self.mode_combo.currentIndexChanged.connect(self._mode_changed) self.approval_bridge.approval_requested.connect(self._show_approval) self.approval_bridge.notice_received.connect(self._show_notice) def _apply_style(self) -> None: self.setStyleSheet( """ QMainWindow, QWidget { background: #171c21; color: #e8edf2; } QLineEdit, QTextEdit, QTextBrowser, QListWidget, QComboBox { background: #202b35; border: 1px solid #344553; border-radius: 6px; padding: 7px; selection-background-color: #b97828; } QPushButton { background: #2b3945; border: 1px solid #485b69; border-radius: 6px; padding: 8px 14px; } QPushButton:hover { border-color: #d3943f; } QPushButton:disabled { color: #68747d; } #panelTitle { color: #d3943f; font-weight: 700; letter-spacing: 1px; } #stateLabel { background: #24342e; color: #86d49b; padding: 10px; border-radius: 6px; font-weight: 700; } QStatusBar { background: #11161a; color: #b8c2ca; } """ ) def _new_session(self) -> None: if self._worker is not None: return session = self.controller.new_session() self._messages = [] self._draft_user = self._draft_assistant = "" self._render_chat() self._refresh_sessions() self.statusBar().showMessage(f"Neue Sitzung: {session.title}", 3000) self.message_edit.setFocus() self._refresh_status() def _open_initial_session(self) -> None: try: sessions = self.controller.list_sessions() if sessions: loaded = self.controller.load_session(sessions[0].id) self._messages = list(loaded.messages) self._render_chat() self._refresh_sessions() self._refresh_status() return except Exception as exc: self._show_error(str(exc)) self._new_session() def _refresh_sessions(self) -> None: try: sessions = self.controller.list_sessions(self.search_edit.text()) except Exception as exc: self._show_error(str(exc)) return active = self.controller.active_session_id self.session_list.clear() for session in sessions: item = QListWidgetItem( f"{session.title}\n" f"{session.last_provider} · {session.message_count} Nachrichten · " f"{session.updated_at[:16].replace('T', ' ')}" ) item.setData(Qt.UserRole, session.id) self.session_list.addItem(item) if session.id == active: self.session_list.setCurrentItem(item) def _load_item(self, item: QListWidgetItem) -> None: if self._worker is not None: return try: loaded = self.controller.load_session(item.data(Qt.UserRole)) except Exception as exc: self._show_error(str(exc)) return self._messages = list(loaded.messages) self._draft_user = self._draft_assistant = "" self._render_chat() self.statusBar().showMessage(f"Sitzung geladen: {loaded.session.title}", 3000) def _session_menu(self, position) -> None: item = self.session_list.itemAt(position) if item is None: return menu = QMenu(self) rename_action = QAction("Umbenennen", menu) rename_action.triggered.connect(lambda: self._rename_item(item)) menu.addAction(rename_action) menu.exec(self.session_list.mapToGlobal(position)) def _rename_item(self, item: QListWidgetItem) -> None: if item.data(Qt.UserRole) != self.controller.active_session_id: self._load_item(item) from PySide6.QtWidgets import QInputDialog title, accepted = QInputDialog.getText(self, "Sitzung umbenennen", "Neuer Titel:") if not accepted: return try: self.controller.rename_active_session(title) self._refresh_sessions() except Exception as exc: self._show_error(str(exc)) def _send(self) -> None: text = self.message_edit.toPlainText().strip() if not text or self._worker is not None: return self.message_edit.clear() self._draft_user = text self._draft_assistant = "" self._render_chat() self._set_generating(True) thread = QThread(self) worker = StreamWorker(self.controller, text) worker.moveToThread(thread) thread.started.connect(worker.run) worker.event_received.connect(self._handle_stream_event) worker.completed.connect(thread.quit) worker.completed.connect(self._generation_complete) thread.finished.connect(worker.deleteLater) thread.finished.connect(thread.deleteLater) self._thread = thread self._worker = worker thread.start() def _handle_stream_event(self, event: StreamEvent) -> None: if event.kind is StreamEventKind.CHUNK: self._draft_assistant += event.text self._render_chat() if event.state is not None: self._set_state(event.state) if event.kind is StreamEventKind.ERROR: self._show_error(event.text) def _generation_complete(self) -> None: failed = self.controller.state in { GenerationState.ABORTED, GenerationState.ERROR, } self._worker = None self._thread = None self._set_generating(False) try: loaded = self.controller.load_session(self.controller.active_session_id or "") self._messages = list(loaded.messages) except Exception as exc: self._show_error(str(exc)) self._draft_user = self._draft_assistant = "" self._render_chat() self._refresh_sessions() self._refresh_status() if failed: self.statusBar().showMessage("Antwort abgebrochen oder fehlgeschlagen.", 5000) else: self.statusBar().showMessage("Antwort vollständig gespeichert.", 3000) def _stop_generation(self) -> None: if self._worker is None: return self._worker.request_cancel() self.approval_bridge.cancel_pending() self.statusBar().showMessage("Abbruch angefordert …") def _set_generating(self, active: bool) -> None: self.send_button.setEnabled(not active) self.stop_button.setEnabled(active) self.new_button.setEnabled(not active) self.session_list.setEnabled(not active) self.mode_combo.setEnabled(not active) def _mode_changed(self) -> None: try: self.controller.set_mode(self.mode_combo.currentData()) self._refresh_status() except Exception as exc: self._show_error(str(exc)) def _show_approval(self, decision: PrivacyDecision) -> None: box = QMessageBox(self) box.setWindowTitle("Cloudfreigabe") box.setIcon(QMessageBox.Question) box.setText( "Diese Anfrage darf nur nach deiner Freigabe an Gemini gesendet werden.\n\n" "Es wird ausschließlich der minimierte notwendige Kontext übertragen." ) box.setInformativeText(f"Lokale Einstufung: {decision.reason}") allow = box.addButton("Einmal erlauben", QMessageBox.AcceptRole) local = box.addButton("Lokal beantworten", QMessageBox.RejectRole) cancel = box.addButton("Abbrechen", QMessageBox.DestructiveRole) box.exec() clicked = box.clickedButton() choice = ApprovalChoice.CANCEL if clicked is allow: choice = ApprovalChoice.ALLOW elif clicked is local: choice = ApprovalChoice.LOCAL elif clicked is cancel: choice = ApprovalChoice.CANCEL self.approval_bridge.resolve(choice) if box.clickedButton() is local: self.statusBar().showMessage("Die Anfrage wird lokal beantwortet.", 3000) def _show_notice(self, text: str) -> None: self.statusBar().showMessage(text, 5000) def _show_error(self, text: str) -> None: QMessageBox.critical(self, "Javis-Fehler", text) self.statusBar().showMessage(f"Fehler: {text}", 5000) def _refresh_status(self) -> None: try: status = self.controller.status() except Exception as exc: self._show_error(str(exc)) return self._set_state(status.state) self.status_labels["provider"].setText(status.provider) self.status_labels["ollama"].setText("Ja" if status.ollama_reachable else "Nein") self.status_labels["model"].setText(status.local_model) self.status_labels["privacy"].setText(status.privacy_policy) self.status_labels["cloud"].setText( str(status.cloud_requests_today) if status.cloud_requests_today is not None else "Unbekannt" ) self.status_labels["fallback"].setText("Ja" if status.local_fallback else "Nein") self.status_labels["data"].setText(status.data_dir) def _set_state(self, state: GenerationState) -> None: self.state_label.setText(state.value) colors = { GenerationState.READY: ("#24342e", "#86d49b"), GenerationState.THINKING: ("#3a3020", "#e1a54f"), GenerationState.ANSWERING: ("#3a3020", "#e1a54f"), GenerationState.ABORTED: ("#3a3020", "#e1a54f"), GenerationState.ERROR: ("#3a2424", "#ed7777"), GenerationState.OFFLINE: ("#3a2424", "#ed7777"), } background, foreground = colors[state] self.state_label.setStyleSheet( f"background: {background}; color: {foreground}; padding: 10px; " "border-radius: 6px; font-weight: 700;" ) def _animate_state(self) -> None: if self.controller.state not in { GenerationState.THINKING, GenerationState.ANSWERING, }: return self._thinking_dots = (self._thinking_dots + 1) % 4 self.state_label.setText(f"{self.controller.state.value}{'.' * self._thinking_dots}") def _render_chat(self) -> None: blocks: list[str] = [] messages = list(self._messages) if self._draft_user: messages.append(ChatMessage("user", self._draft_user)) if self._draft_assistant: messages.append(ChatMessage("assistant", self._draft_assistant)) for message in messages: label = "Pascal" if message.role == "user" else "Javis" content = message.content if message.role == "assistant" else escape(message.content) blocks.append(f"### {label}\n\n{content}") self.chat_view.setMarkdown("\n\n---\n\n".join(blocks)) scrollbar = self.chat_view.verticalScrollBar() scrollbar.setValue(scrollbar.maximum()) def _focus_search(self) -> None: self.search_edit.setFocus() self.search_edit.selectAll() def closeEvent(self, event: QCloseEvent) -> None: self._stop_generation() if self._thread is not None and not self._thread.wait(2500): event.ignore() self.statusBar().showMessage( "Javis beendet die laufende Antwort noch kontrolliert.", 3000, ) return event.accept()