Coverage for src / lilbee / cli / tui / app.py: 100%
169 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-04-29 19:16 +0000
1"""Main Textual app for lilbee TUI."""
3from __future__ import annotations
5import contextlib
6import logging
7from collections.abc import Callable
8from pathlib import Path
9from typing import Any, ClassVar
11from textual.app import App, ComposeResult
12from textual.binding import Binding, BindingType
13from textual.css.query import NoMatches
14from textual.screen import Screen
15from textual.signal import Signal
17from lilbee import settings
18from lilbee.cli.tui import messages as msg
19from lilbee.cli.tui.commands import LilbeeCommandProvider
20from lilbee.cli.tui.widgets.status_bar import ViewTabs
21from lilbee.config import cfg
22from lilbee.services import get_services, reset_services
24log = logging.getLogger(__name__)
26_DEFAULT_THEME = "gruvbox" # warm retro CRT aesthetic
27_CHAT_SCREEN_NAME = "chat"
28DARK_THEMES = (
29 "monokai",
30 "dracula",
31 "tokyo-night",
32 "nord",
33 "gruvbox",
34 "catppuccin-mocha",
35 "catppuccin-frappe",
36 "atom-one-dark",
37 "rose-pine",
38 "solarized-dark",
39 "textual-dark",
40)
43def _make_catalog() -> Screen:
44 from lilbee.cli.tui.screens.catalog import CatalogScreen
46 return CatalogScreen()
49def _make_status() -> Screen:
50 from lilbee.cli.tui.screens.status import StatusScreen
52 return StatusScreen()
55def _make_settings() -> Screen:
56 from lilbee.cli.tui.screens.settings import SettingsScreen
58 return SettingsScreen()
61def _make_tasks() -> Screen:
62 from lilbee.cli.tui.screens.task_center import TaskCenter
64 return TaskCenter()
67def _make_wiki() -> Screen:
68 from lilbee.cli.tui.screens.wiki import WikiScreen
70 return WikiScreen()
73_BASE_VIEWS: dict[str, Callable[[], Screen]] = {
74 "Catalog": _make_catalog,
75 "Status": _make_status,
76 "Settings": _make_settings,
77 "Tasks": _make_tasks,
78}
81def get_views() -> dict[str, Callable[[], Screen]]:
82 """Return the active view factories, including wiki when enabled."""
83 views = dict(_BASE_VIEWS)
84 if cfg.wiki:
85 views["Wiki"] = _make_wiki
86 return views
89def _on_settings_changed_evict_cache(payload: tuple[str, object]) -> None:
90 """Drop loaded-model state when a load-affecting setting changes."""
91 # Lazy: llama_cpp_provider's transitive imports cost ~500ms.
92 from lilbee.providers.llama_cpp_provider import LOAD_AFFECTING_KEYS
94 key, _value = payload
95 if key in LOAD_AFFECTING_KEYS:
96 get_services().provider.invalidate_load_cache()
99class LilbeeApp(App[None]):
100 """Full-screen TUI for lilbee knowledge base."""
102 TITLE = "lilbee"
103 CSS_PATH = Path(__file__).parent / "app.tcss"
104 ENABLE_COMMAND_PALETTE = True
105 COMMANDS = {LilbeeCommandProvider} # noqa: RUF012
107 _NAV_GROUP = Binding.Group("Navigate")
109 BINDINGS: ClassVar[list[BindingType]] = [
110 Binding("question_mark", "push_help", "Help", show=True),
111 Binding("f1", "push_help", "Help", show=False),
112 Binding("ctrl+h", "push_help", "Help", show=False),
113 Binding("ctrl+t", "cycle_theme", "Theme", show=True),
114 Binding("t", "open_tasks", "Tasks", show=True),
115 # priority=True is required: even though NavAwareInput lets [ and ]
116 # bubble past Input.check_consume_key, Textual's focused Input still
117 # handles printable keys in _on_key before a non-priority ancestor
118 # binding can fire. Both NavAwareInput and priority=True are needed.
119 Binding(
120 "left_square_bracket",
121 "nav_prev",
122 "Prev",
123 show=True,
124 group=_NAV_GROUP,
125 priority=True,
126 ),
127 Binding(
128 "right_square_bracket",
129 "nav_next",
130 "Next",
131 show=True,
132 group=_NAV_GROUP,
133 priority=True,
134 ),
135 Binding("ctrl+c", "quit", "Quit", show=True, priority=True),
136 ]
138 def __init__(self, *, auto_sync: bool = False, initial_view: str | None = None) -> None:
139 super().__init__()
140 self._auto_sync = auto_sync
141 self._initial_view = initial_view
142 self.active_view = msg.DEFAULT_VIEW
143 self._switching = False
144 self._theme_index = 0
145 self.last_quit_time: float = 0.0
146 self.settings_changed_signal: Signal[tuple[str, object]] = Signal(self, "settings_changed")
147 from lilbee.cli.tui.widgets.task_bar import TaskBarController
149 self.task_bar = TaskBarController(self)
151 def compose(self) -> ComposeResult:
152 yield from () # screens compose their own ViewTabs + Footer
154 def on_mount(self) -> None:
155 self.title = f"lilbee — {cfg.chat_model}"
156 # Restore the persisted theme so the TUI opens in whatever the user
157 # picked last session, not always the gruvbox default.
158 persisted = cfg.theme or _DEFAULT_THEME
159 self.theme = persisted if persisted in self.available_themes else _DEFAULT_THEME
160 self._sync_theme_index_to_current()
162 self.settings_changed_signal.subscribe(self, _on_settings_changed_evict_cache)
164 from lilbee.cli.tui.screens.chat import ChatScreen
166 chat = ChatScreen(auto_sync=self._auto_sync)
167 self.install_screen(chat, name=_CHAT_SCREEN_NAME)
168 self.push_screen(_CHAT_SCREEN_NAME)
169 if self._initial_view and self._initial_view != msg.DEFAULT_VIEW:
170 self.switch_view(self._initial_view)
172 def action_cycle_theme(self) -> None:
173 self._theme_index = (self._theme_index + 1) % len(DARK_THEMES)
174 name = DARK_THEMES[self._theme_index]
175 self._apply_and_persist_theme(name)
176 self.notify(msg.THEME_SET.format(name=name))
178 def set_theme(self, name: str) -> None:
179 """Set theme by name (used by /theme command). Persists across sessions."""
180 if name in self.available_themes:
181 self._apply_and_persist_theme(name)
182 self._sync_theme_index_to_current()
184 def _apply_and_persist_theme(self, name: str) -> None:
185 """Apply *name* live and write it to config.toml."""
186 self.theme = name
187 cfg.theme = name
188 settings.set_value(cfg.data_root, "theme", name)
190 def set_active_model(self, key: str, value: str) -> None:
191 """Single write boundary for active model refs; persists the
192 post-validator value so subscribers see the normalized form."""
193 setattr(cfg, key, value)
194 normalized = getattr(cfg, key)
195 settings.set_value(cfg.data_root, key, normalized)
196 self.settings_changed_signal.publish((key, normalized))
198 def _sync_theme_index_to_current(self) -> None:
199 """Align the cycle index with the active theme so Ctrl+T moves from there."""
200 try:
201 self._theme_index = DARK_THEMES.index(self.theme)
202 except ValueError:
203 self._theme_index = 0
205 async def action_quit(self) -> None:
206 """Context-aware Ctrl+C: cancel active task > cancel stream > quit.
207 On second Ctrl+C (within 2s), force-exits via os._exit to handle
208 cases where the GIL is held by native code.
209 """
210 import time
212 now = time.monotonic()
213 if now - self.last_quit_time < 2.0:
214 self._force_quit()
215 return
216 self.last_quit_time = now
218 if not self.task_bar.queue.is_empty:
219 active = self.task_bar.queue.active_task
220 if active:
221 self.task_bar.cancel_task(active.task_id)
222 self.notify(msg.APP_CANCELLED)
223 return
224 from lilbee.cli.tui.screens.chat import ChatScreen
225 from lilbee.cli.tui.screens.setup import SetupWizard
227 screen = self.screen
228 if isinstance(screen, SetupWizard):
229 screen.action_cancel()
230 return
231 if isinstance(screen, ChatScreen) and screen.streaming:
232 screen.action_cancel_stream()
233 return
234 self.exit()
236 def _force_quit(self) -> None:
237 """Force-exit when normal quit is blocked (e.g. GIL held by native code)."""
238 import os
240 with contextlib.suppress(Exception):
241 reset_services()
242 os._exit(1)
244 def switch_view(self, view_name: str) -> None:
245 """Switch to a named view via lazy screen factories.
247 Guards against concurrent switches: ``switch_screen`` is async
248 (processed on the next event-loop tick) but callers read
249 ``active_view`` synchronously. Without a guard, rapid keypresses
250 queue conflicting switches that corrupt the screen stack.
251 ``active_view`` is updated after the switch completes.
252 """
253 if self._switching:
254 return
255 self._switching = True
257 if view_name == "Chat":
258 from lilbee.cli.tui.screens.chat import ChatScreen
260 if not isinstance(self.screen, ChatScreen):
261 self.switch_screen(_CHAT_SCREEN_NAME)
262 # Already on Chat, just update state below.
263 else:
264 factory = get_views().get(view_name)
265 if factory is None:
266 self._switching = False
267 return
268 self.switch_screen(factory())
270 def _finish() -> None:
271 self.active_view = view_name
272 self._switching = False
273 # ViewTabs.on_mount captured active_view before this callback
274 # runs, so the highlight would lag by one step without this push.
275 with contextlib.suppress(NoMatches):
276 self.screen.query_one(ViewTabs).active_view = view_name
278 self.call_later(_finish)
280 def action_push_help(self) -> None:
281 if self.screen.query("HelpPanel"):
282 self.action_hide_help_panel()
283 else:
284 self.action_show_help_panel()
286 def action_open_tasks(self) -> None:
287 """Jump to the Task Center screen (t key)."""
288 self.switch_view("Tasks")
290 def action_nav_prev(self) -> None:
291 """Navigate to previous view ([ key)."""
292 view_names = msg.get_nav_views()
293 current_idx = view_names.index(self.active_view)
294 self.switch_view(view_names[(current_idx - 1) % len(view_names)])
296 def action_nav_next(self) -> None:
297 """Navigate to next view (] key)."""
298 view_names = msg.get_nav_views()
299 current_idx = view_names.index(self.active_view)
300 self.switch_view(view_names[(current_idx + 1) % len(view_names)])
303def apply_active_model(host_app: App[Any], key: str, value: str) -> None:
304 """Route model writes through LilbeeApp.set_active_model; bare-App fallback for tests."""
305 if isinstance(host_app, LilbeeApp):
306 host_app.set_active_model(key, value)
307 return
308 setattr(cfg, key, value)
309 settings.set_value(cfg.data_root, key, getattr(cfg, key))