Coverage for src / lilbee / cli / tui / app.py: 100%
257 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-06-28 01:01 +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.app.settings import apply_settings_update
20from lilbee.app.themes import DARK_THEMES
21from lilbee.cli.tui import messages as msg
22from lilbee.cli.tui.commands import LilbeeCommandProvider
23from lilbee.cli.tui.widgets.status_bar import ViewTabs
24from lilbee.core.config import cfg
25from lilbee.providers.worker.transport import WorkerRole
27log = logging.getLogger(__name__)
29_DEFAULT_THEME = "rose-pine" # muted, low-glare; easier on the eyes than the warmer themes
30_CHAT_SCREEN_NAME = "chat"
31# Long enough that a model-fallback notice is readable before it fades.
32_FALLBACK_TOAST_TIMEOUT_S = 10.0
35def _view_screen_name(view_name: str) -> str:
36 """Stable install_screen identifier for a top-level view (lower-cased)."""
37 return view_name.lower()
40def _make_catalog() -> Screen:
41 from lilbee.cli.tui.screens.catalog import CatalogScreen
43 return CatalogScreen()
46def _make_status() -> Screen:
47 from lilbee.cli.tui.screens.status import StatusScreen
49 return StatusScreen()
52def _make_settings() -> Screen:
53 from lilbee.cli.tui.screens.settings import SettingsScreen
55 return SettingsScreen()
58def _make_tasks() -> Screen:
59 from lilbee.cli.tui.screens.task_center import TaskCenter
61 return TaskCenter()
64def _make_wiki() -> Screen:
65 from lilbee.cli.tui.screens.wiki import WikiScreen
67 return WikiScreen()
70_BASE_VIEWS: dict[str, Callable[[], Screen]] = {
71 "Catalog": _make_catalog,
72 "Status": _make_status,
73 "Settings": _make_settings,
74 "Tasks": _make_tasks,
75}
78def get_views() -> dict[str, Callable[[], Screen]]:
79 """Return the active view factories, including wiki when enabled."""
80 views = dict(_BASE_VIEWS)
81 if cfg.wiki:
82 views["Wiki"] = _make_wiki
83 return views
86class LilbeeApp(App[None]):
87 """Full-screen TUI for lilbee knowledge base."""
89 TITLE = "lilbee"
90 CSS_PATH = Path(__file__).parent / "app.tcss"
91 ENABLE_COMMAND_PALETTE = True
92 COMMANDS = {LilbeeCommandProvider} # noqa: RUF012
94 _NAV_GROUP = Binding.Group("Navigate")
96 BINDINGS: ClassVar[list[BindingType]] = [
97 # ``?`` is non-priority so a focused TextArea (chat input in INSERT
98 # mode) can swallow it and type the literal character. F1 / Ctrl+H
99 # remain priority routes that always open help, even mid-typing.
100 Binding("question_mark", "push_help", "Help", show=False),
101 Binding("f1", "push_help", "Help", show=True, priority=True),
102 Binding("ctrl+h", "push_help", "Help", show=False, priority=True),
103 Binding("escape", "dismiss_help_if_open", "Close help", show=False, priority=True),
104 Binding("ctrl+t", "cycle_theme", "Theme", show=True),
105 Binding("t", "open_tasks", "Tasks", show=True),
106 Binding("f4", "toggle_lilbee_path", "Path/Name", show=True),
107 # Non-priority so Chat's "focus_commands" and Catalog's
108 # "focus_search" still win on those screens. Fires only on
109 # screens that don't bind slash themselves, routing the user
110 # to Chat with the slash already typed.
111 Binding("slash", "global_slash_to_chat", "Command", show=False),
112 # priority=True so a focused TextArea cannot swallow the bracket
113 # under stress (multi-key send-keys etc.); type literal brackets
114 # via Shift+[ / Shift+] which produce { / } and bypass these.
115 Binding(
116 "left_square_bracket",
117 "nav_prev",
118 "Prev",
119 show=True,
120 group=_NAV_GROUP,
121 priority=True,
122 ),
123 Binding(
124 "right_square_bracket",
125 "nav_next",
126 "Next",
127 show=True,
128 group=_NAV_GROUP,
129 priority=True,
130 ),
131 Binding("ctrl+c", "quit", "Quit", show=True, priority=True),
132 Binding("S", "run_sync", "Sync", show=False, priority=True),
133 ]
135 def __init__(self, *, initial_view: str | None = None) -> None:
136 super().__init__()
137 self._initial_view = initial_view
138 self.active_view = msg.DEFAULT_VIEW
139 self._switching = False
140 self._theme_index = 0
141 # Names of non-Chat screens already installed via install_screen.
142 # Subsequent visits switch by name to reuse the same instance,
143 # so Footer / signal / worker wiring runs once per session.
144 self._installed_screen_names: set[str] = set()
145 self.settings_changed_signal: Signal[tuple[str, object]] = Signal(self, "settings_changed")
146 self.provider_availability_changed_signal: Signal[tuple[str, object]] = Signal(
147 self, "provider_availability_changed"
148 )
149 from lilbee.cli.tui.widgets.task_bar_controller import TaskBarController
151 self.task_bar = TaskBarController(self)
153 def compose(self) -> ComposeResult:
154 yield from () # screens compose their own ViewTabs + Footer
156 # Test seam: the TUI test fixtures subclass LilbeeApp and set this to True
157 # so on_mount short-circuits before the heavyweight setup (model
158 # canonicalization, ChatScreen install, signal subscriptions, sync probe).
159 # Production never sets it. See tests/_lilbee_app_test_host.py.
160 _test_skip_auto_init: ClassVar[bool] = False
162 def on_mount(self) -> None:
163 if self._test_skip_auto_init:
164 return
165 self._canonicalize_persisted_models()
166 self.title = f"lilbee: {cfg.chat_model}"
167 # Restore the persisted theme so the TUI opens in whatever the user
168 # picked last session, not always the default.
169 persisted = cfg.theme or _DEFAULT_THEME
170 self.theme = persisted if persisted in self.available_themes else _DEFAULT_THEME
171 self._sync_theme_index_to_current()
173 self.settings_changed_signal.subscribe(self, self._fan_out_provider_availability)
174 self._wire_worker_pool_notifications()
176 from lilbee.cli.tui.screens.chat import ChatScreen
178 chat = ChatScreen()
179 self.install_screen(chat, name=_CHAT_SCREEN_NAME)
180 self.push_screen(_CHAT_SCREEN_NAME)
181 if self._initial_view and self._initial_view != msg.DEFAULT_VIEW:
182 self.switch_view(self._initial_view)
183 # Cheap detection only: filesystem walk + hash compare. The user
184 # initiates sync explicitly via S or the command palette.
185 self.task_bar.start_detect_pending()
187 def _wire_worker_pool_notifications(self) -> None:
188 """Surface worker spawn lifecycle in the bottom TaskBar.
190 Worker spawns happen on the pool runtime thread, not the TUI's main
191 loop, so the listeners marshal back via :meth:`call_from_thread`
192 before mutating controller state. A single TaskBar hint covers all
193 in-flight roles instead of one toast per role; the chat surface is
194 for user content, not implementation detail.
195 """
197 def _on_spawning(role: WorkerRole) -> None:
198 self.call_from_thread(self.task_bar.mark_role_spawning, role.value)
200 def _on_spawned(role: WorkerRole) -> None:
201 self.call_from_thread(self.task_bar.mark_role_spawned, role.value)
203 get_services().add_pool_listener(on_spawning=_on_spawning, on_spawned=_on_spawned)
205 def _canonicalize_persisted_models(self) -> None:
206 """Swap stale persisted refs to a working fallback, persist, and log once."""
207 from lilbee.modelhub.model_manager import (
208 ValidationResult,
209 canonicalize_chat_model,
210 canonicalize_embedding_model,
211 )
213 for canon, field, label in (
214 (canonicalize_chat_model(), "chat_model", "Chat"),
215 (canonicalize_embedding_model(), "embedding_model", "Embedding"),
216 ):
217 if canon.status == ValidationResult.OK:
218 continue
219 reason = canon.reason or msg.MODEL_REASON_DEFAULT
221 if canon.original == canon.effective:
222 # Nothing to fall back to: keep the ref; the chat screen opens the wizard.
223 notice = msg.MODEL_UNUSABLE_OPENING_SETUP.format(
224 label=label, original=canon.original, reason=reason
225 )
226 log.warning(notice)
227 self.notify(notice, severity="warning", timeout=_FALLBACK_TOAST_TIMEOUT_S)
228 continue
230 # A rejected swap (validation or disk error) must not be fatal at startup.
231 try:
232 apply_settings_update({field: canon.effective})
233 except (ValueError, OSError):
234 log.warning(
235 msg.MODEL_FALLBACK_FAILED.format(
236 label=label,
237 original=canon.original,
238 effective=canon.effective,
239 reason=reason,
240 ),
241 exc_info=True,
242 )
243 continue
244 notice = msg.MODEL_FALLBACK_NOTICE.format(
245 label=label, original=canon.original, effective=canon.effective, reason=reason
246 )
247 log.warning(notice)
248 self.notify(notice, severity="warning", timeout=_FALLBACK_TOAST_TIMEOUT_S)
250 def _fan_out_provider_availability(self, payload: tuple[str, object]) -> None:
251 """Republish on provider_availability_changed_signal when an API key changes."""
252 from lilbee.core.config.keys import PROVIDER_API_KEYS
254 key, value = payload
255 if key in PROVIDER_API_KEYS:
256 self.provider_availability_changed_signal.publish((key, value))
258 def action_cycle_theme(self) -> None:
259 self._theme_index = (self._theme_index + 1) % len(DARK_THEMES)
260 name = DARK_THEMES[self._theme_index]
261 self._apply_and_persist_theme(name)
262 self.notify(msg.THEME_SET.format(name=name))
264 def action_toggle_lilbee_path(self) -> None:
265 """Flip the status-bar pill between the friendly name and the data-root path."""
266 self.set_setting("show_lilbee_path", not cfg.show_lilbee_path)
268 def set_theme(self, name: str) -> None:
269 """Set theme by name (used by /theme command). Persists across sessions."""
270 if name in self.available_themes:
271 self._apply_and_persist_theme(name)
272 self._sync_theme_index_to_current()
274 def _apply_and_persist_theme(self, name: str) -> None:
275 """Apply *name* live and write it to config.toml."""
277 self.theme = name
278 apply_settings_update({"theme": name})
280 def set_active_model(self, key: str, value: str) -> None:
281 """Persist an active model ref through the shared write boundary.
283 Refs whose download is still queued or active are refused before the
284 boundary runs, so a half-pulled file cannot land in a model slot.
285 """
287 downloading = self.task_bar.downloading_label_for(value)
288 if downloading is not None:
289 self.notify(
290 msg.MODEL_BEING_DOWNLOADED.format(name=downloading),
291 severity="warning",
292 )
293 return
294 try:
295 apply_settings_update({key: value})
296 except ValueError as exc:
297 self.notify(msg.MODEL_ASSIGN_REJECTED.format(error=exc), severity="error")
298 return
299 self.settings_changed_signal.publish((key, getattr(cfg, key)))
301 def set_setting(self, key: str, value: object) -> None:
302 """Apply a writable / model-role setting through the boundary, then fan out to the UI.
304 Raises ``ValueError`` for keys outside ``WRITABLE_CONFIG_FIELDS | MODEL_ROLE_FIELDS``
305 or values rejected by pydantic validation. Callers either catch and toast or let it
306 propagate.
307 """
309 apply_settings_update({key: value})
310 normalized = getattr(cfg, key)
311 if key == "theme" and isinstance(normalized, str) and normalized in self.available_themes:
312 self.theme = normalized
313 self._sync_theme_index_to_current()
314 self.settings_changed_signal.publish((key, normalized))
316 def _sync_theme_index_to_current(self) -> None:
317 """Align cycle index with the active theme."""
318 try:
319 self._theme_index = DARK_THEMES.index(self.theme)
320 except ValueError:
321 self._theme_index = 0
323 async def action_quit(self) -> None:
324 """Context-aware Ctrl+C: cancel active task > cancel stream > quit."""
325 get_services().cancel_inference()
327 if not self.task_bar.queue.is_empty:
328 active = self.task_bar.queue.active_task
329 if active:
330 self.task_bar.cancel_task(active.task_id)
331 self.notify(msg.APP_CANCELLED)
332 return
333 from lilbee.cli.tui.screens.chat import ChatScreen
334 from lilbee.cli.tui.screens.setup import SetupWizard
336 screen = self.screen
337 if isinstance(screen, SetupWizard):
338 screen.action_cancel()
339 return
340 if isinstance(screen, ChatScreen) and screen.streaming:
341 screen.action_cancel_stream()
342 return
343 self.exit()
345 def switch_view(self, view_name: str) -> None:
346 """Switch to a named view, installing each screen at most once.
348 Guards against concurrent switches via ``self._switching`` so
349 rapid keypresses don't corrupt the screen stack.
350 ``active_view`` is updated after the switch completes.
351 """
352 if self._switching:
353 return
354 self._switching = True
356 if view_name == "Chat":
357 from lilbee.cli.tui.screens.chat import ChatScreen
359 if not isinstance(self.screen, ChatScreen):
360 self.switch_screen(_CHAT_SCREEN_NAME)
361 # Already on Chat, just update state below.
362 else:
363 factory = get_views().get(view_name)
364 if factory is None:
365 self._switching = False
366 return
367 screen_name = _view_screen_name(view_name)
368 if screen_name not in self._installed_screen_names:
369 self.install_screen(factory(), name=screen_name)
370 self._installed_screen_names.add(screen_name)
371 self.switch_screen(screen_name)
373 def _finish() -> None:
374 self.active_view = view_name
375 self._switching = False
376 # ViewTabs.on_mount captured active_view before this callback
377 # runs, so the highlight would lag by one step without this push.
378 with contextlib.suppress(NoMatches):
379 self.screen.query_one(ViewTabs).active_view = view_name
381 self.call_later(_finish)
383 def action_push_help(self) -> None:
384 if self.screen.query("HelpPanel"):
385 self.action_hide_help_panel()
386 else:
387 self.action_show_help_panel()
389 def action_command_palette(self) -> None:
390 """Ctrl+P: cycle the chat dropdown if visible, else open the palette."""
391 from lilbee.cli.tui.screens.chat import ChatScreen
392 from lilbee.cli.tui.widgets.autocomplete import CompletionOverlay
394 screen = self.screen
395 if isinstance(screen, ChatScreen):
396 try:
397 overlay = screen.query_one("#completion-overlay", CompletionOverlay)
398 except NoMatches:
399 overlay = None
400 if overlay is not None and overlay.is_visible:
401 screen.action_complete_prev()
402 return
403 super().action_command_palette()
405 def action_dismiss_help_if_open(self) -> None:
406 """Esc dismisses the HelpPanel when it is open; otherwise no-op.
408 Without this, focus inside the panel could prevent ``?`` from
409 toggling it back off and the user had no key to escape with.
410 Bubble the Escape so screens can still receive it when no panel
411 is mounted.
412 """
413 from textual.actions import SkipAction
415 if self.screen.query("HelpPanel"):
416 self.action_hide_help_panel()
417 return
418 raise SkipAction()
420 def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:
421 """Hide ``t Tasks`` from the footer while a text input is focused.
423 ``t`` is not a priority binding, so a focused ``Input`` / ``TextArea``
424 (the chat prompt in INSERT mode, a catalog/settings search box) eats
425 it as a literal character. Showing ``t Tasks`` there would lie.
426 """
427 # isinstance: a focused Input/TextArea consumes printable keys before
428 # non-priority screen/app bindings see them, so `t` types a literal there.
429 if action == "open_tasks" and isinstance(self.focused, (Input, TextArea)):
430 return False
431 return super().check_action(action, parameters)
433 def action_open_tasks(self) -> None:
434 """Jump to the Task Center screen (t key)."""
435 self.switch_view("Tasks")
437 def action_global_slash_to_chat(self) -> None:
438 """Route a slash typed on a non-slash-bound screen back to Chat's prompt.
440 Lets the user type ``/setup`` from Settings/Tasks/etc. without
441 the next character (``s``, ``t``, ...) hitting a global single-key
442 binding before the slash command can compose.
443 """
444 from lilbee.cli.tui.screens.chat import ChatScreen
446 if not isinstance(self.screen, ChatScreen):
447 self.switch_view("Chat")
448 # Defer the prompt focus until after switch_view's call_later
449 # _finish has updated active_view, so the chat input is mounted
450 # and ready when we prefill it.
451 self.call_later(self._prefill_chat_command)
453 def _prefill_chat_command(self) -> None:
454 """Focus the chat input and seed it with a leading slash."""
455 from lilbee.cli.tui.screens.chat import ChatScreen
457 if isinstance(self.screen, ChatScreen):
458 self.screen.action_focus_commands()
460 def action_run_sync(self) -> None:
461 """Trigger an explicit document sync from any screen (S key).
463 The TaskBar hint is rendered globally, so the trigger must work
464 everywhere. Routes to the registered ChatScreen which owns the
465 ``_run_sync`` orchestration; switches to the Chat view first if
466 not already there so the user can watch progress.
467 """
468 from lilbee.cli.tui.screens.chat import ChatScreen
470 if isinstance(self.screen, ChatScreen):
471 self.screen._run_sync()
472 return
473 try:
474 chat = self.get_screen(_CHAT_SCREEN_NAME, ChatScreen)
475 except KeyError:
476 return
477 self.switch_view("Chat")
479 def _start() -> None:
480 if isinstance(self.screen, ChatScreen):
481 chat._run_sync()
483 self.call_later(_start)
485 def action_nav_prev(self) -> None:
486 """Navigate to previous view ([ key)."""
487 view_names = msg.get_nav_views()
488 current_idx = view_names.index(self.active_view)
489 self.switch_view(view_names[(current_idx - 1) % len(view_names)])
491 def action_nav_next(self) -> None:
492 """Navigate to next view (] key)."""
493 view_names = msg.get_nav_views()
494 current_idx = view_names.index(self.active_view)
495 self.switch_view(view_names[(current_idx + 1) % len(view_names)])
498def apply_active_model(host_app: App[Any], key: str, value: str) -> None:
499 """Route model writes through LilbeeApp.set_active_model."""
500 cast(LilbeeApp, host_app).set_active_model(key, value)
503def apply_setting(host_app: App[Any], key: str, value: object) -> None:
504 """Route non-model settings writes through LilbeeApp.set_setting."""
505 cast(LilbeeApp, host_app).set_setting(key, value)