"""Textual TUI application for Navi Code.""" from __future__ import annotations from textual.app import App, ComposeResult from textual.containers import Horizontal from textual.widgets import Footer, Header from clients.terminal import api from clients.terminal.config import settings from clients.terminal.state import StateManager from clients.terminal.tui.commands.registry import get_registry from clients.terminal.tui.context import TuiContext from clients.terminal.tui.events import ( ConnectionStatusChanged, PermissionRequest, UserSubmitted, WsEvent, ) from clients.terminal.tui.permissions import PermissionEngine from clients.terminal.tui.widgets import ChatPanel, InputBox, StatusPanel from clients.terminal.tui.ws_bridge import WsBridge class NaviCodeTui(App): """OpenCode-inspired terminal UI for Navi.""" CSS = """ Screen { align: center middle; } NaviCodeTui { padding: 0; } Header { height: 1; } Footer { height: 1; } """ BINDINGS = [ ("ctrl+p", "command_palette", "Palette"), ("ctrl+x q", "quit", "Quit"), ("ctrl+x n", "new_session", "New"), ("ctrl+x l", "list_sessions", "Sessions"), ("ctrl+x c", "compact", "Compact"), ("ctrl+x t", "toggle_thinking", "Thinking"), ] def __init__( self, session_id: str | None = None, profile_id: str | None = None, new_session: bool = False, ) -> None: super().__init__() self._chat_panel = ChatPanel() self._status_panel = StatusPanel() self._input_box = InputBox() self._state = StateManager() self._ctx = TuiContext( state=self._state, chat_panel=self._chat_panel, status_panel=self._status_panel, ) self._bridge: WsBridge | None = None self._permission_engine = PermissionEngine() self._pending_permission: PermissionRequest | None = None self._requested_session_id = session_id self._requested_profile_id = profile_id self._force_new_session = new_session def compose(self) -> ComposeResult: yield Header(show_clock=False) with Horizontal(): yield self._chat_panel yield self._status_panel yield self._input_box yield Footer() def on_mount(self) -> None: self.run_worker(self._startup) async def _startup(self) -> None: session_id = await self._resolve_session( self._requested_session_id, self._requested_profile_id, self._force_new_session, ) if session_id: await self._attach_session(session_id) self._input_box.focus_input() async def _resolve_session( self, session_id: str | None, profile_id: str | None, force_new: bool, ) -> str | None: if session_id and not force_new: try: session = api.get_session(session_id) return session["session_id"] except Exception: pass if not force_new: saved = self._state.get_session_id() if saved: try: session = api.get_session(saved) return session["session_id"] except Exception: self._state.clear_session_id() profile = profile_id or settings.default_profile_id try: session = api.create_session(profile) except Exception as exc: self._chat_panel.handle_ws_event({"type": "error", "message": f"Failed to create session: {exc}"}) return None self._state.set_session_id(session["session_id"]) return session["session_id"] async def _attach_session(self, session_id: str) -> None: self._ctx.session_id = session_id try: session = api.get_session(session_id) self._ctx.profile_id = session.get("profile_id") or settings.default_profile_id except Exception: self._ctx.profile_id = settings.default_profile_id self._status_panel.set_session(session_id) self._status_panel.set_profile(self._ctx.profile_id) self._status_panel.set_model(settings.ollama_default_model if hasattr(settings, "ollama_default_model") else "unknown") if self._bridge: await self._bridge.stop() self._bridge = WsBridge(self, session_id) await self._bridge.start() self._ctx.ws_client = self._bridge.client self._chat_panel.handle_ws_event({"type": "status", "content": f"Connected to {session_id[:8]}"}) def on_user_submitted(self, event: UserSubmitted) -> None: text = event.text if text.startswith("/"): self._run_command(text) return # @ file references and ! shell commands can be parsed here in Phase 4. self._chat_panel.add_user_message(text) if self._bridge and self._bridge.connected: self._bridge.client.enqueue(text) else: self._chat_panel.handle_ws_event({"type": "error", "message": "Not connected to a session"}) def _run_command(self, text: str) -> None: parts = text[1:].split(None, 1) name = parts[0].lower() args = parts[1] if len(parts) > 1 else "" registry = get_registry() cmd = registry.get(name) if cmd is None: self._chat_panel.handle_ws_event({"type": "error", "message": f"Unknown command: /{name}"}) return self.run_worker(self._command_worker(cmd, args)) async def _command_worker(self, cmd, args: str) -> None: await cmd.execute(self._ctx, args) def on_ws_event(self, event: WsEvent) -> None: self._chat_panel.handle_ws_event(event.payload) def on_connection_status_changed(self, event: ConnectionStatusChanged) -> None: self._status_panel.set_connection(event.connected, event.detail) def on_permission_request(self, event: PermissionRequest) -> None: self._pending_permission = event self._chat_panel.handle_ws_event( { "type": "status", "content": f"Permission required: {event.details}\nAllow once (y) / always (a) / reject (n)", } ) def action_command_palette(self) -> None: # Phase 4: implement command palette screen. self._chat_panel.handle_ws_event({"type": "status", "content": "Command palette: /help, /new, /sessions, /switch, /profile, /thinking, /compact, /quit"}) def action_new_session(self) -> None: self._run_command("/new") def action_list_sessions(self) -> None: self._run_command("/sessions") def action_compact(self) -> None: self._run_command("/compact") def action_toggle_thinking(self) -> None: self._run_command("/thinking") async def action_quit(self) -> None: if self._bridge: await self._bridge.stop() self.exit() async def on_unmount(self) -> None: if self._bridge: await self._bridge.stop()