import asyncio import os import ollama import sys import re import pty import threading import signal import fcntl import select from textual.app import App, ComposeResult from textual.containers import Vertical, Horizontal from textual.widgets import Header, Footer, RichLog, Input, Static, Button from textual.binding import Binding from textual.message import Message from textual import log # --- NACHRICHTEN FÜR DIE KOMMUNIKATION --- class ConnectionStatus(Message): """Meldet das Ergebnis des Ollama-Verbindungstests.""" def __init__(self, success: bool, error: Exception | None) -> None: self.success = success self.error = error super().__init__() class PermissionPrompt(Vertical): def compose(self) -> ComposeResult: yield Static("Die KI möchte einen Befehl ausführen. Erlauben?", id="permission_question") yield RichLog(id="permission_command_display", wrap=True, markup=True, classes="command-box") with Horizontal(classes="permission-buttons"): yield Button("Ausführen (Ja)", variant="success", id="perm_yes") yield Button("Ignorieren (Nein)", variant="error", id="perm_no") yield Button("Immer ausführen", variant="primary", id="perm_always") class ChatView(Vertical): """Der obere Bereich für den Chat.""" def compose(self) -> ComposeResult: log.info("ChatView: compose()") yield Static("Ollama (qwen3-coder:latest)", classes="box-title") yield RichLog(id="chat_log", wrap=True, markup=True) yield PermissionPrompt(id="permission_prompt", classes="hidden") yield Input(placeholder="Nachricht an die KI senden... (Tab wechselt Fokus)", id="chat_input") def on_mount(self) -> None: log.info("ChatView: on_mount()") self.query_one("#chat_input", Input).focus() def write_message(self, message: str) -> None: self.query_one("#chat_log", RichLog).write(message) class ConsoleView(Vertical): """Der untere Bereich für die Shell.""" def compose(self) -> ComposeResult: log.info("ConsoleView: compose()") yield Static("Pseudo-Terminal (PTY)", classes="box-title") yield RichLog(id="console_log", wrap=True, markup=True) yield Input(placeholder="Shell-Befehl eingeben...", id="console_input") def write_message(self, message: str) -> None: self.query_one("#console_log", RichLog).write(message) class AICompanionApp(App): TITLE = "Ollama Split-Shell Companion" CSS = """ Screen { layout: vertical; } .box-title { background: $accent; color: $text; padding-left: 1; height: 1; text-style: bold; } ChatView { height: 50%; border-bottom: solid $primary; } ConsoleView { height: 50%; } #chat_log, #console_log { height: 1fr; padding: 0 1; } #chat_log { background: $surface-darken-1; } #console_log { background: black; color: white; scrollbar-color: $primary; } Input { dock: bottom; height: 3; border-top: solid $secondary; } Input:focus { border-top: solid $accent; } .hidden { display: none; } #permission_prompt { background: $panel; border: round $secondary; padding: 1; height: auto; margin-top: 1; } #permission_question { margin-bottom: 1; } .command-box { border: solid $primary; padding: 1; height: auto; max-height: 10; } .permission-buttons { height: auto; align: center middle; margin-top: 1; } .permission-buttons Button { margin: 0 1; } """ BINDINGS = [Binding("ctrl+c,ctrl+q", "quit", "Beenden", show=True, priority=True)] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.chat_history = [{'role': 'system', 'content': "Du bist ein hilfreicher Coding-Assistent..."}] self.console_context_buffer, self.permission_mode, self.pending_command = "", "ask", None self.MODEL_NAME, self.OLLAMA_HOST = "qwen3-coder:latest", 'http://ai:11434' self.ollama_client = ollama.AsyncClient(host=self.OLLAMA_HOST) self.pty_master_fd, self.pty_pid, self.pty_reader_thread = None, None, None self.pty_thread_running = False # --- KORREKTUR: Die entscheidende, fehlende Methode --- def compose(self) -> ComposeResult: """Erstellt das Layout der App und fügt die Kind-Widgets hinzu.""" log.info("AICompanionApp: compose()") yield Header() yield ChatView() yield ConsoleView() yield Footer() def on_mount(self) -> None: """Wird aufgerufen, nachdem `compose` fertig ist. Jetzt können wir die Widgets finden.""" log.info("AICompanionApp: on_mount()") # Speichern der Referenzen auf die Kind-Widgets für späteren Gebrauch self.chat_view = self.query_one(ChatView) self.console_view = self.query_one(ConsoleView) # Initiale Nachrichten anzeigen self.chat_view.write_message(f"[yellow]Verbinde mit Ollama unter {self.OLLAMA_HOST}...[/yellow]") self.console_view.write_message("[bold green]$[/bold green] Warte auf Ollama-Verbindung...") # Worker starten self.run_worker(self.test_ollama_connection_worker, name="OllamaConnector", thread=True) def on_unmount(self) -> None: log.info("AICompanionApp: on_unmount()") self.cleanup_pty_process() def test_ollama_connection_worker(self) -> None: """Läuft in einem Thread, um die UI nicht zu blockieren.""" log.info("OllamaConnector Worker: Starting connection test.") try: sync_client = ollama.Client(host=self.OLLAMA_HOST) sync_client.list() log.info("OllamaConnector Worker: Connection successful.") self.post_message(ConnectionStatus(success=True, error=None)) except Exception as e: log.info(f"OllamaConnector Worker: Connection failed. Error: {e}") self.post_message(ConnectionStatus(success=False, error=e)) async def on_connection_status(self, message: ConnectionStatus) -> None: """Empfängt die Nachricht vom Worker und aktualisiert die UI sicher.""" log.info(f"AICompanionApp: Received ConnectionStatus(success={message.success})") if message.success: self.chat_view.write_message("[green]Verbindung mit Ollama erfolgreich hergestellt.[/green]") self.console_view.write_message(f"[bold green]$[/bold green] Bereit. Ollama verbunden.") else: self.chat_view.write_message(f"[bold red]Verbindung zu Ollama fehlgeschlagen:[/bold red] {message.error}") self.console_view.write_message(f"[bold red]$[/bold red] FEHLER: Ollama nicht erreichbar.") # ... Der Rest des Codes bleibt unverändert ... async def on_input_submitted(self, message: Input.Submitted) -> None: if message.input.id == "chat_input": await self.handle_chat_submission(message.value) elif message.input.id == "console_input": await self.handle_console_submission(message.value) async def on_button_pressed(self, event: Button.Pressed) -> None: permission_prompt = self.query_one(PermissionPrompt) if event.button.id == "perm_yes": self.execute_pending_command() elif event.button.id == "perm_always": self.permission_mode = "always"; self.execute_pending_command() self.chat_view.write_message("[italic blue]Hinweis: Ab jetzt werden alle Befehle automatisch ausgeführt.[/italic blue]") permission_prompt.add_class("hidden") self.query_one("#chat_input", Input).remove_class("hidden") self.query_one("#chat_input", Input).focus() self.pending_command = None def execute_pending_command(self): if self.pending_command: console_input = self.query_one("#console_input", Input) console_input.value = self.pending_command self.call_later(self.handle_console_submission, self.pending_command) console_input.focus() def parse_and_handle_commands(self, text: str): commands = re.findall(r"```(?:bash|sh)?\n(.*?)\n```", text, re.DOTALL) if not commands: return False self.pending_command = commands[0].strip() if self.permission_mode == "always": self.chat_view.write_message(f"[italic blue]Führe Befehl automatisch aus: [bold]{self.pending_command}[/bold][/italic blue]") self.execute_pending_command(); return True self.query_one("#permission_command_display", RichLog).clear() self.query_one("#permission_command_display", RichLog).write(self.pending_command) self.query_one(PermissionPrompt).remove_class("hidden") self.query_one("#chat_input", Input).add_class("hidden") self.query_one("#perm_yes", Button).focus() return True async def handle_chat_submission(self, user_input: str) -> None: chat_input = self.query_one("#chat_input", Input) if not user_input.strip(): return self.chat_view.write_message(f"[bold green]USER>[/bold green] {user_input}") chat_input.value = "" if self.console_context_buffer: self.chat_history.append({'role': 'user', 'content': f"SYSTEM (Konsolen-Output):\n```\n{self.console_context_buffer}\n```"}) self.console_context_buffer = "" self.chat_history.append({'role': 'user', 'content': user_input}) try: stream = await self.ollama_client.chat(model=self.MODEL_NAME, messages=self.chat_history, stream=True) ai_response, live_widget = "", Static(f"[bold cyan]{self.MODEL_NAME}>[/bold cyan] ", classes="live-response") await self.chat_view.mount(live_widget) async for chunk in stream: content = chunk['message']['content']; ai_response += content live_widget.update(f"[bold cyan]{self.MODEL_NAME}>[/bold cyan] {ai_response}") self.chat_history.append({'role': 'assistant', 'content': ai_response}) live_widget.remove() self.chat_view.write_message(f"[bold cyan]{self.MODEL_NAME}>[/bold cyan] {ai_response}") self.parse_and_handle_commands(ai_response) except Exception as e: self.chat_view.write_message(f"\n[bold red]Fehler bei Ollama ({self.MODEL_NAME}):[/bold red] {e}\n") async def handle_console_submission(self, command: str) -> None: console_input = self.query_one("#console_input", Input); console_input.value = "" if self.pty_pid and self.pty_master_fd: os.write(self.pty_master_fd, (command + "\n").encode()); return if not command.strip(): return self.cleanup_pty_process() self.console_view.write_message(f"[bold green]$ {command}[/bold green]") try: pid, master_fd = pty.fork() except OSError as e: self.console_view.write_message(f"[bold red]Fehler beim Erstellen des PTY: {e}[/bold red]"); return if pid == 0: shell = os.environ.get("SHELL", "/bin/bash"); os.execv(shell, [shell, "-c", command]) else: self.pty_pid, self.pty_master_fd = pid, master_fd self.pty_thread_running = True self.pty_reader_thread = threading.Thread(target=self._pty_reader_loop, daemon=True) self.pty_reader_thread.start() def _pty_reader_loop(self): while self.pty_thread_running and self.pty_master_fd: try: if select.select([self.pty_master_fd], [], [], 0.1)[0]: data = os.read(self.pty_master_fd, 1024) if data: self.call_from_thread(self.update_console, data.decode(errors='replace')) else: break except (OSError, select.error): break self.call_from_thread(self.cleanup_pty_process) def update_console(self, data: str) -> None: self.console_view.write_message(data) self.console_context_buffer += data if len(self.console_context_buffer) > 4000: self.console_context_buffer = self.console_context_buffer[-4000:] def cleanup_pty_process(self): if self.pty_thread_running: self.pty_thread_running = False if self.pty_reader_thread and self.pty_reader_thread.is_alive(): self.pty_reader_thread.join(timeout=0.2) if self.pty_master_fd: try: os.close(self.pty_master_fd) except OSError: pass self.pty_master_fd = None if self.pty_pid: try: os.kill(self.pty_pid, signal.SIGKILL); os.waitpid(self.pty_pid, 0) except (OSError, ProcessLookupError): pass self.pty_pid = None if self.is_mounted and not self._unmounted: try: self.console_view.write_message("[bold green]$[/bold green] ") except Exception: pass if __name__ == "__main__": app = AICompanionApp() app.run()