Coverage for src / lilbee / cli / tui / app.py: 100%
259 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-05-15 20:55 +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, cast
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
16from textual.widgets import Input, TextArea
18from lilbee.app.services import get_services
19from lilbee.cli.tui import messages as msg
20from lilbee.cli.tui.commands import LilbeeCommandProvider
21from lilbee.cli.tui.widgets.status_bar import ViewTabs
22from lilbee.core import settings
23from lilbee.core.config import cfg, validate_model_task_assignment
24from lilbee.providers.worker.transport import WorkerRole
26log = logging.getLogger(__name__)
28_DEFAULT_THEME = "rose-pine" # muted, low-glare; easier on the eyes than the warmer themes
29_CHAT_SCREEN_NAME = "chat"
30DARK_THEMES = (
31 "monokai",
32 "dracula",
33 "tokyo-night",
34 "nord",
35 "gruvbox",
36 "catppuccin-mocha",
37 "catppuccin-frappe",
38 "atom-one-dark",
39 "rose-pine",
40 "solarized-dark",
41 "textual-dark",
42)
45def _view_screen_name(view_name: str) -> str:
46 """Stable install_screen identifier for a top-level view (lower-cased)."""
47 return view_name.lower()
50def _make_catalog() -> Screen:
51 from lilbee.cli.tui.screens.catalog import CatalogScreen
53 return CatalogScreen()
56def _make_status() -> Screen:
57 from lilbee.cli.tui.screens.status import StatusScreen
59 return StatusScreen()
62def _make_settings() -> Screen:
63 from lilbee.cli.tui.screens.settings import SettingsScreen
65 return SettingsScreen()
68def _make_tasks() -> Screen:
69 from lilbee.cli.tui.screens.task_center import TaskCenter
71 return TaskCenter()
74def _make_wiki() -> Screen:
75 from lilbee.cli.tui.screens.wiki import WikiScreen
77 return WikiScreen()
80_BASE_VIEWS: dict[str, Callable[[], Screen]] = {
81 "Catalog": _make_catalog,
82 "Status": _make_status,
83 "Settings": _make_settings,
84 "Tasks": _make_tasks,
85}
88def get_views() -> dict[str, Callable[[], Screen]]:
89 """Return the active view factories, including wiki when enabled."""
90 views = dict(_BASE_VIEWS)
91 if cfg.wiki:
92 views["Wiki"] = _make_wiki
93 return views
96_MODEL_REF_KEYS = frozenset({"chat_model", "embedding_model", "vision_model", "reranker_model"})
99def _on_settings_changed_evict_cache(payload: tuple[str, object]) -> None:
100 """Drop loaded-model state when a load-affecting setting changes."""
101 from lilbee.providers.llama_cpp.provider import (
102 LOAD_AFFECTING_KEYS,
103 PER_CALL_RELOADABLE_KEYS,
104 )
106 key, _value = payload
107 if key in LOAD_AFFECTING_KEYS and key not in PER_CALL_RELOADABLE_KEYS:
108 # Roles that do NOT honor per-call request.model (embed, rerank, plus
109 # any role-agnostic key like num_ctx) need the pool to drop the worker
110 # so the next call respawns under the new cfg. Chat and vision workers
111 # observe the new path on the next request via _ensure_loaded and
112 # reload in place, saving the 1-3 s spawn cost.
113 get_services().provider.invalidate_load_cache()
114 if key in _MODEL_REF_KEYS:
115 from lilbee.modelhub.model_info import invalidate_cache
117 invalidate_cache()
120class LilbeeApp(App[None]):
121 """Full-screen TUI for lilbee knowledge base."""
123 TITLE = "lilbee"
124 CSS_PATH = Path(__file__).parent / "app.tcss"
125 ENABLE_COMMAND_PALETTE = True
126 COMMANDS = {LilbeeCommandProvider} # noqa: RUF012
128 _NAV_GROUP = Binding.Group("Navigate")
130 BINDINGS: ClassVar[list[BindingType]] = [
131 # ``?`` is non-priority so a focused TextArea (chat input in INSERT
132 # mode) can swallow it and type the literal character. F1 / Ctrl+H
133 # remain priority routes that always open help, even mid-typing.
134 Binding("question_mark", "push_help", "Help", show=False),
135 Binding("f1", "push_help", "Help", show=True, priority=True),
136 Binding("ctrl+h", "push_help", "Help", show=False, priority=True),
137 Binding("escape", "dismiss_help_if_open", "Close help", show=False, priority=True),
138 Binding("ctrl+t", "cycle_theme", "Theme", show=True),
139 Binding("t", "open_tasks", "Tasks", show=True),
140 # Non-priority so Chat's "focus_commands" and Catalog's
141 # "focus_search" still win on those screens. Fires only on
142 # screens that don't bind slash themselves, routing the user
143 # to Chat with the slash already typed.
144 Binding("slash", "global_slash_to_chat", "Command", show=False),
145 # priority=True so a focused TextArea cannot swallow the bracket
146 # under stress (multi-key send-keys etc.); type literal brackets
147 # via Shift+[ / Shift+] which produce { / } and bypass these.
148 Binding(
149 "left_square_bracket",
150 "nav_prev",
151 "Prev",
152 show=True,
153 group=_NAV_GROUP,
154 priority=True,
155 ),
156 Binding(
157 "right_square_bracket",
158 "nav_next",
159 "Next",
160 show=True,
161 group=_NAV_GROUP,
162 priority=True,
163 ),
164 Binding("ctrl+c", "quit", "Quit", show=True, priority=True),
165 Binding("S", "run_sync", "Sync", show=False, priority=True),
166 ]
168 def __init__(self, *, initial_view: str | None = None) -> None:
169 super().__init__()
170 self._initial_view = initial_view
171 self.active_view = msg.DEFAULT_VIEW
172 self._switching = False
173 self._theme_index = 0
174 # Names of non-Chat screens already installed via install_screen.
175 # Subsequent visits switch by name to reuse the same instance,
176 # so Footer / signal / worker wiring runs once per session.
177 self._installed_screen_names: set[str] = set()
178 self.settings_changed_signal: Signal[tuple[str, object]] = Signal(self, "settings_changed")
179 self.provider_availability_changed_signal: Signal[tuple[str, object]] = Signal(
180 self, "provider_availability_changed"
181 )
182 from lilbee.cli.tui.widgets.task_bar_controller import TaskBarController
184 self.task_bar = TaskBarController(self)
186 def compose(self) -> ComposeResult:
187 yield from () # screens compose their own ViewTabs + Footer
189 # Test seam: the TUI test fixtures subclass LilbeeApp and set this to True
190 # so on_mount short-circuits before the heavyweight setup (model
191 # canonicalization, ChatScreen install, signal subscriptions, sync probe).
192 # Production never sets it. See tests/_lilbee_app_test_host.py.
193 _test_skip_auto_init: ClassVar[bool] = False
195 def on_mount(self) -> None:
196 if self._test_skip_auto_init:
197 return
198 self._canonicalize_persisted_models()
199 self.title = f"lilbee: {cfg.chat_model}"
200 # Restore the persisted theme so the TUI opens in whatever the user
201 # picked last session, not always the default.
202 persisted = cfg.theme or _DEFAULT_THEME
203 self.theme = persisted if persisted in self.available_themes else _DEFAULT_THEME
204 self._sync_theme_index_to_current()
206 self.settings_changed_signal.subscribe(self, _on_settings_changed_evict_cache)
207 self.settings_changed_signal.subscribe(self, self._fan_out_provider_availability)
208 self._wire_worker_pool_notifications()
210 from lilbee.cli.tui.screens.chat import ChatScreen
212 chat = ChatScreen()
213 self.install_screen(chat, name=_CHAT_SCREEN_NAME)
214 self.push_screen(_CHAT_SCREEN_NAME)
215 if self._initial_view and self._initial_view != msg.DEFAULT_VIEW:
216 self.switch_view(self._initial_view)
217 # Cheap detection only: filesystem walk + hash compare. The user
218 # initiates sync explicitly via S or the command palette.
219 self.task_bar.start_detect_pending()
221 def _wire_worker_pool_notifications(self) -> None:
222 """Surface worker spawn lifecycle in the bottom TaskBar.
224 Worker spawns happen on the pool runtime thread, not the TUI's main
225 loop, so the listeners marshal back via :meth:`call_from_thread`
226 before mutating controller state. A single TaskBar hint covers all
227 in-flight roles instead of one toast per role; the chat surface is
228 for user content, not implementation detail.
229 """
231 def _on_spawning(role: WorkerRole) -> None:
232 self.call_from_thread(self.task_bar.mark_role_spawning, role.value)
234 def _on_spawned(role: WorkerRole) -> None:
235 self.call_from_thread(self.task_bar.mark_role_spawned, role.value)
237 get_services().add_pool_listener(on_spawning=_on_spawning, on_spawned=_on_spawned)
239 def _canonicalize_persisted_models(self) -> None:
240 """Swap stale persisted refs to a working fallback, persist, and log once.
242 Persisting the swap via ``settings.set_value`` is what makes this a
243 one-time notice. The previous version only updated cfg in memory, so
244 the warning fired every restart for as long as the stale ref sat in
245 ``config.toml``.
246 """
247 from lilbee.modelhub.model_manager import (
248 ValidationResult,
249 canonicalize_chat_model,
250 canonicalize_embedding_model,
251 )
253 for canon, field, label in (
254 (canonicalize_chat_model(), "chat_model", "Chat"),
255 (canonicalize_embedding_model(), "embedding_model", "Embedding"),
256 ):
257 if canon.status == ValidationResult.OK or canon.original == canon.effective:
258 continue
259 setattr(cfg, field, canon.effective)
260 settings.set_value(cfg.data_root, field, canon.effective)
261 log.warning(
262 msg.MODEL_FALLBACK_NOTICE.format(
263 label=label, original=canon.original, effective=canon.effective
264 )
265 )
267 def _fan_out_provider_availability(self, payload: tuple[str, object]) -> None:
268 """Republish on provider_availability_changed_signal when an API key changes."""
269 from lilbee.core.config.keys import PROVIDER_API_KEYS
271 key, value = payload
272 if key in PROVIDER_API_KEYS:
273 self.provider_availability_changed_signal.publish((key, value))
275 def action_cycle_theme(self) -> None:
276 self._theme_index = (self._theme_index + 1) % len(DARK_THEMES)
277 name = DARK_THEMES[self._theme_index]
278 self._apply_and_persist_theme(name)
279 self.notify(msg.THEME_SET.format(name=name))
281 def set_theme(self, name: str) -> None:
282 """Set theme by name (used by /theme command). Persists across sessions."""
283 if name in self.available_themes:
284 self._apply_and_persist_theme(name)
285 self._sync_theme_index_to_current()
287 def _apply_and_persist_theme(self, name: str) -> None:
288 """Apply *name* live and write it to config.toml."""
289 self.theme = name
290 cfg.theme = name
291 settings.set_value(cfg.data_root, "theme", name)
293 def set_active_model(self, key: str, value: str) -> None:
294 """Single write boundary for active model refs.
296 Validates the ref's catalog task matches the field, so a chat-only
297 model cannot land in the embedding slot (and equivalents for vision /
298 rerank). Provider-prefixed refs and the empty string pass through.
299 """
300 try:
301 canonical = validate_model_task_assignment(key, value)
302 except ValueError as exc:
303 self.notify(msg.MODEL_ASSIGN_REJECTED.format(error=exc), severity="error")
304 return
305 setattr(cfg, key, canonical)
306 normalized = getattr(cfg, key)
307 settings.set_value(cfg.data_root, key, normalized)
308 self.settings_changed_signal.publish((key, normalized))
310 def set_setting(self, key: str, value: object) -> None:
311 """Single write boundary for non-model settings."""
312 setattr(cfg, key, value)
313 normalized = getattr(cfg, key)
314 # settings.set_value persists into TOML, which only accepts strings.
315 # Persisting None as "" used to break pydantic-settings load (no
316 # coercion from "" to int|None), so drop the key on None and let
317 # the field default apply on next startup.
318 if normalized is None:
319 settings.delete_value(cfg.data_root, key)
320 else:
321 if isinstance(normalized, list):
322 persisted = "\n".join(str(x) for x in normalized)
323 else:
324 persisted = str(normalized)
325 settings.set_value(cfg.data_root, key, persisted)
326 if key == "theme" and isinstance(normalized, str) and normalized in self.available_themes:
327 self.theme = normalized
328 self._sync_theme_index_to_current()
329 self.settings_changed_signal.publish((key, normalized))
331 def _sync_theme_index_to_current(self) -> None:
332 """Align cycle index with the active theme."""
333 try:
334 self._theme_index = DARK_THEMES.index(self.theme)
335 except ValueError:
336 self._theme_index = 0
338 async def action_quit(self) -> None:
339 """Context-aware Ctrl+C: cancel active task > cancel stream > quit."""
340 get_services().cancel_inference()
342 if not self.task_bar.queue.is_empty:
343 active = self.task_bar.queue.active_task
344 if active:
345 self.task_bar.cancel_task(active.task_id)
346 self.notify(msg.APP_CANCELLED)
347 return
348 from lilbee.cli.tui.screens.chat import ChatScreen
349 from lilbee.cli.tui.screens.setup import SetupWizard
351 screen = self.screen
352 if isinstance(screen, SetupWizard):
353 screen.action_cancel()
354 return
355 if isinstance(screen, ChatScreen) and screen.streaming:
356 screen.action_cancel_stream()
357 return
358 self.exit()
360 def switch_view(self, view_name: str) -> None:
361 """Switch to a named view, installing each screen at most once.
363 Guards against concurrent switches via ``self._switching`` so
364 rapid keypresses don't corrupt the screen stack.
365 ``active_view`` is updated after the switch completes.
366 """
367 if self._switching:
368 return
369 self._switching = True
371 if view_name == "Chat":
372 from lilbee.cli.tui.screens.chat import ChatScreen
374 if not isinstance(self.screen, ChatScreen):
375 self.switch_screen(_CHAT_SCREEN_NAME)
376 # Already on Chat, just update state below.
377 else:
378 factory = get_views().get(view_name)
379 if factory is None:
380 self._switching = False
381 return
382 screen_name = _view_screen_name(view_name)
383 if screen_name not in self._installed_screen_names:
384 self.install_screen(factory(), name=screen_name)
385 self._installed_screen_names.add(screen_name)
386 self.switch_screen(screen_name)
388 def _finish() -> None:
389 self.active_view = view_name
390 self._switching = False
391 # ViewTabs.on_mount captured active_view before this callback
392 # runs, so the highlight would lag by one step without this push.
393 with contextlib.suppress(NoMatches):
394 self.screen.query_one(ViewTabs).active_view = view_name
396 self.call_later(_finish)
398 def action_push_help(self) -> None:
399 if self.screen.query("HelpPanel"):
400 self.action_hide_help_panel()
401 else:
402 self.action_show_help_panel()
404 def action_command_palette(self) -> None:
405 """Ctrl+P: cycle the chat dropdown if visible, else open the palette."""
406 from lilbee.cli.tui.screens.chat import ChatScreen
407 from lilbee.cli.tui.widgets.autocomplete import CompletionOverlay
409 screen = self.screen
410 if isinstance(screen, ChatScreen):
411 try:
412 overlay = screen.query_one("#completion-overlay", CompletionOverlay)
413 except NoMatches:
414 overlay = None
415 if overlay is not None and overlay.is_visible:
416 overlay.cycle_prev()
417 return
418 super().action_command_palette()
420 def action_dismiss_help_if_open(self) -> None:
421 """Esc dismisses the HelpPanel when it is open; otherwise no-op.
423 Without this, focus inside the panel could prevent ``?`` from
424 toggling it back off and the user had no key to escape with.
425 Bubble the Escape so screens can still receive it when no panel
426 is mounted.
427 """
428 from textual.actions import SkipAction
430 if self.screen.query("HelpPanel"):
431 self.action_hide_help_panel()
432 return
433 raise SkipAction()
435 def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:
436 """Hide ``t Tasks`` from the footer while a text input is focused.
438 ``t`` is not a priority binding, so a focused ``Input`` / ``TextArea``
439 (the chat prompt in INSERT mode, a catalog/settings search box) eats
440 it as a literal character. Showing ``t Tasks`` there would lie.
441 """
442 # isinstance: a focused Input/TextArea consumes printable keys before
443 # non-priority screen/app bindings see them, so `t` types a literal there.
444 if action == "open_tasks" and isinstance(self.focused, (Input, TextArea)):
445 return False
446 return super().check_action(action, parameters)
448 def action_open_tasks(self) -> None:
449 """Jump to the Task Center screen (t key)."""
450 self.switch_view("Tasks")
452 def action_global_slash_to_chat(self) -> None:
453 """Route a slash typed on a non-slash-bound screen back to Chat's prompt.
455 Lets the user type ``/setup`` from Settings/Tasks/etc. without
456 the next character (``s``, ``t``, ...) hitting a global single-key
457 binding before the slash command can compose.
458 """
459 from lilbee.cli.tui.screens.chat import ChatScreen
461 if not isinstance(self.screen, ChatScreen):
462 self.switch_view("Chat")
463 # Defer the prompt focus until after switch_view's call_later
464 # _finish has updated active_view, so the chat input is mounted
465 # and ready when we prefill it.
466 self.call_later(self._prefill_chat_command)
468 def _prefill_chat_command(self) -> None:
469 """Focus the chat input and seed it with a leading slash."""
470 from lilbee.cli.tui.screens.chat import ChatScreen
472 if isinstance(self.screen, ChatScreen):
473 self.screen.action_focus_commands()
475 def action_run_sync(self) -> None:
476 """Trigger an explicit document sync from any screen (S key).
478 The TaskBar hint is rendered globally, so the trigger must work
479 everywhere. Routes to the registered ChatScreen which owns the
480 ``_run_sync`` orchestration; switches to the Chat view first if
481 not already there so the user can watch progress.
482 """
483 from lilbee.cli.tui.screens.chat import ChatScreen
485 if isinstance(self.screen, ChatScreen):
486 self.screen._run_sync()
487 return
488 try:
489 chat = self.get_screen(_CHAT_SCREEN_NAME, ChatScreen)
490 except KeyError:
491 return
492 self.switch_view("Chat")
494 def _start() -> None:
495 if isinstance(self.screen, ChatScreen):
496 chat._run_sync()
498 self.call_later(_start)
500 def action_nav_prev(self) -> None:
501 """Navigate to previous view ([ key)."""
502 view_names = msg.get_nav_views()
503 current_idx = view_names.index(self.active_view)
504 self.switch_view(view_names[(current_idx - 1) % len(view_names)])
506 def action_nav_next(self) -> None:
507 """Navigate to next view (] key)."""
508 view_names = msg.get_nav_views()
509 current_idx = view_names.index(self.active_view)
510 self.switch_view(view_names[(current_idx + 1) % len(view_names)])
513def apply_active_model(host_app: App[Any], key: str, value: str) -> None:
514 """Route model writes through LilbeeApp.set_active_model."""
515 cast(LilbeeApp, host_app).set_active_model(key, value)
518def apply_setting(host_app: App[Any], key: str, value: object) -> None:
519 """Route non-model settings writes through LilbeeApp.set_setting."""
520 cast(LilbeeApp, host_app).set_setting(key, value)