Coverage for src/lilbee/cli/tui/screens/status.py: 100%
275 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Status screen: knowledge base info with collapsible sections."""
3from __future__ import annotations
5import asyncio
6import contextlib
7import logging
8from dataclasses import dataclass, field
9from pathlib import Path
10from typing import TYPE_CHECKING, ClassVar
12if TYPE_CHECKING:
13 from lilbee.cli.tui.app import LilbeeApp
15from rich.text import Text
16from textual import work
17from textual.app import ComposeResult
18from textual.binding import Binding, BindingType
19from textual.containers import VerticalScroll
20from textual.content import Content
21from textual.screen import Screen
22from textual.widgets import Collapsible, DataTable, Static
23from textual.worker import Worker, WorkerState
25from lilbee.app.services import get_services
26from lilbee.app.settings import SCANNED_PAGES_LABEL, scanned_pages_state
27from lilbee.app.status import held_out_sources
28from lilbee.cli.tui import messages as msg
29from lilbee.cli.tui.browse_bindings import BROWSE_LIST_BINDINGS, browse_back_bindings
30from lilbee.cli.tui.pill import pill
31from lilbee.core.config import cfg
32from lilbee.core.text import strip_control_characters
33from lilbee.data.store import DOCUMENTS_PAGE_DEFAULT, SourceRecord
34from lilbee.data.types import SkippedSource
35from lilbee.modelhub.model_info import ModelArchInfo, get_model_architecture
37log = logging.getLogger(__name__)
39# Rows per page of the Documents table. Must exceed the table's visible height
40# (``max-height`` in status.tcss) or the table never scrolls and never asks for more.
41_DOCS_PAGE_SIZE = DOCUMENTS_PAGE_DEFAULT
42# Load the next page when the table is scrolled this far down, or when the cursor
43# reaches this far into the loaded rows. A cursor move scrolls the table only after
44# its layout settles, so the cursor carries its own trigger.
45_DOCS_SCROLL_PREFETCH_RATIO = 0.85
46_WORKER_FETCH_SOURCES = "status_fetch_sources"
47_WORKER_FETCH_DOCS_PAGE = "status_fetch_docs_page"
50@dataclass
51class _DocsResult:
52 """Outcome of the background sources read.
54 ``load_failed`` distinguishes "store opened but empty" (``sources == []``)
55 from "the read raised" so the UI shows the right placeholder.
56 """
58 sources: list[SourceRecord]
59 load_failed: bool
60 document_count: int = 0
61 held_out: list[SkippedSource] = field(default_factory=list)
62 held_out_total: int = 0
65@dataclass
66class _DocsPage:
67 """One further page of sources, or the fact that reading it failed."""
69 sources: list[SourceRecord]
70 load_failed: bool = False
73def _model_pill(name: str) -> Content:
74 """Return a green 'loaded' pill if name is set, red 'not set' otherwise."""
75 if name:
76 return pill("loaded", "$success", "$text")
77 return pill("not set", "$error", "$text")
80# Label-column width used across the status sections so keys line up
81# when scanned vertically. Values past this column render bold.
82_KV_LABEL_WIDTH = 14
85def _kv_line(label: str, value: str | Content, status: Content | None = None) -> Content:
86 """Assemble one key/value row: dim padded label, bold value, optional pill."""
87 padded = label.ljust(_KV_LABEL_WIDTH)
88 parts: list[Content] = [Content.styled(padded, "$text-muted")]
89 if isinstance(value, Content):
90 parts.append(value)
91 else:
92 parts.append(Content.styled(value, "bold"))
93 if status is not None:
94 parts.append(Content(" "))
95 parts.append(status)
96 return Content.assemble(*parts)
99def _collapse_home(path: Path | str) -> str:
100 """Replace the user's home prefix with '~' so long paths stay scannable."""
101 text = str(path)
102 home = str(Path.home())
103 return text.replace(home, "~", 1) if text.startswith(home) else text
106def _data_dir_pill() -> Content:
107 """Return a pill based on whether the data directory exists."""
108 if Path(cfg.data_dir).exists():
109 return pill("exists", "$success", "$text")
110 return pill("missing", "$error", "$text")
113def _build_config_content() -> Content:
114 """Build the configuration section content."""
115 lines = [
116 _kv_line("Data dir", _collapse_home(cfg.data_dir), _data_dir_pill()),
117 _kv_line("Chat model", cfg.chat_model or "(disabled)", _model_pill(cfg.chat_model)),
118 _kv_line(
119 "Embed model", cfg.embedding_model or "(disabled)", _model_pill(cfg.embedding_model)
120 ),
121 _kv_line("Vision model", cfg.vision_model or "(disabled)", _model_pill(cfg.vision_model)),
122 _kv_line("Reranker", cfg.reranker_model or "(disabled)", _model_pill(cfg.reranker_model)),
123 _kv_line(SCANNED_PAGES_LABEL, scanned_pages_state()),
124 ]
125 return Content("\n").join(lines)
128def _build_storage_content(doc_count: int) -> Content:
129 """Build the storage section content."""
130 lines = [
131 _kv_line("Documents", str(doc_count)),
132 _kv_line("Data dir", _collapse_home(cfg.data_dir)),
133 _kv_line("Models dir", _collapse_home(cfg.models_dir)),
134 ]
135 if cfg.entity_extraction:
136 from lilbee.app.status import entity_status
138 section = entity_status()
139 if section is not None:
140 names = ", ".join(section.types) or "schema pending"
141 lines.append(_kv_line("Entities", f"{section.rows} extracted ({names})"))
142 return Content("\n").join(lines)
145def _build_arch_content(info: ModelArchInfo) -> Content:
146 """Build the model architecture section from GGUF metadata."""
147 lines = [
148 _kv_line("Chat arch", info.chat_arch),
149 _kv_line("Embed arch", info.embed_arch),
150 _kv_line("Handler", pill(info.active_handler, "$accent", "$text")),
151 ]
152 if info.vision_projector:
153 lines.append(_kv_line("Vision proj", info.vision_projector))
154 return Content("\n").join(lines)
157class StatusScreen(Screen[None]):
158 """Knowledge base status view with collapsible sections."""
160 app: LilbeeApp # type: ignore[assignment]
162 CSS_PATH = "status.tcss"
163 AUTO_FOCUS = "CollapsibleTitle"
164 HELP = (
165 "Knowledge base status.\n\n"
166 "View configuration, documents, model architecture, and storage info."
167 )
169 BINDINGS: ClassVar[list[BindingType]] = [
170 *browse_back_bindings(),
171 Binding("tab", "app.focus_next", "Next section", show=False),
172 Binding(
173 "shift+tab",
174 "app.focus_previous",
175 "Prev section",
176 show=False,
177 ),
178 *BROWSE_LIST_BINDINGS,
179 ]
181 def __init__(self) -> None:
182 super().__init__()
183 self._sections_mounted: bool = False
184 self._pending_docs: _DocsResult | None = None
185 self._pending_arch: ModelArchInfo | None = None
186 self._docs_loaded: int = 0
187 self._docs_total: int = 0
188 self._docs_has_more: bool = False
189 self._docs_loading: bool = False
190 self._docs_page_failed: bool = False
191 self._docs_scroll_watched: bool = False
193 def compose(self) -> ComposeResult:
194 from textual.widgets import Footer
196 from lilbee.cli.tui.widgets.bottom_bars import BottomBars
197 from lilbee.cli.tui.widgets.status_bar import ViewTabs
198 from lilbee.cli.tui.widgets.task_bar import TaskBar
199 from lilbee.cli.tui.widgets.top_bars import TopBars
201 with TopBars():
202 yield ViewTabs()
203 # Mount only the first (Configuration) collapsible up front so the
204 # screen paints fast on push. Documents/arch/storage hydrate via
205 # ``call_after_refresh`` once the screen is visible -- their
206 # backing widgets are still cheap to mount, but the synchronous
207 # cost of mounting all four under a single VerticalScroll spiked
208 # screen-switch latency to ~1s on cold caches.
209 yield VerticalScroll(
210 Collapsible(
211 Static(id="config-info"),
212 title="Configuration",
213 id="config-section",
214 collapsed=False,
215 ),
216 id="status-scroll",
217 )
218 with BottomBars():
219 yield TaskBar()
220 yield Footer()
222 def on_mount(self) -> None:
223 # ``cfg`` reads are in-memory and cheap. Anything that touches
224 # disk runs in a worker so the screen paints instantly.
225 # ``get_model_architecture`` opens up to three GGUF files and
226 # parses their headers (~hundreds of ms each cold); ``get_sources``
227 # reads LanceDB (seconds on cold caches).
228 self._load_config()
229 self.call_after_refresh(self._mount_remaining_sections)
230 self._fetch_sources_worker()
231 self._fetch_arch_worker()
233 async def _mount_remaining_sections(self) -> None:
234 """Mount Documents/Architecture/Storage once the screen is visible."""
235 if not self.is_mounted:
236 return
237 scroll = self.query_one("#status-scroll", VerticalScroll)
238 await scroll.mount_all(
239 [
240 Collapsible(
241 DataTable(id="docs-table"),
242 Static(id="docs-more"),
243 title=msg.STATUS_DOCS_TITLE,
244 id="docs-section",
245 collapsed=False,
246 ),
247 Collapsible(
248 DataTable(id="held-out-table"),
249 title=msg.STATUS_HELD_OUT_TITLE,
250 id="held-out-section",
251 collapsed=True,
252 ),
253 Collapsible(
254 Static(id="arch-info"),
255 title="Model Architecture",
256 id="arch-section",
257 collapsed=False,
258 ),
259 Collapsible(
260 Static(id="storage-info"),
261 title="Storage",
262 id="storage-section",
263 collapsed=False,
264 ),
265 ]
266 )
267 self._sections_mounted = True
268 # Yield once so Textual gets a chance to compose the freshly-
269 # mounted Collapsibles' children. Without this, querying
270 # #docs-table immediately after mount_all races on Windows.
271 await asyncio.sleep(0)
272 self._show_loading_placeholders()
273 # Replay any worker callbacks that arrived before the deferred
274 # mount completed.
275 if self._pending_docs is not None:
276 self._apply_docs(self._pending_docs)
277 self._pending_docs = None
278 if self._pending_arch is not None:
279 self._load_arch(self._pending_arch)
280 self._pending_arch = None
282 def _show_loading_placeholders(self) -> None:
283 """Surface a 'Loading…' marker for sections backed by workers.
285 Wrapped in NoMatches suppression because Collapsible children
286 compose on the next refresh tick, which on Windows can outlast
287 the synchronous return from mount_all. Worker callbacks repaint
288 the same widgets when they arrive, so a missed placeholder is
289 only a brief cosmetic gap.
290 """
291 from textual.css.query import NoMatches
293 with contextlib.suppress(NoMatches):
294 table = self.query_one("#docs-table", DataTable)
295 table.add_columns("Document", "Chunks")
296 table.cursor_type = "row"
297 table.add_row("Loading...", "")
298 with contextlib.suppress(NoMatches):
299 self.query_one("#storage-info", Static).update(
300 Content.styled("Loading...", "$text-muted")
301 )
302 with contextlib.suppress(NoMatches):
303 self.query_one("#arch-info", Static).update(Content.styled("Loading...", "$text-muted"))
305 @work(thread=True, name=_WORKER_FETCH_SOURCES, exit_on_error=False)
306 def _fetch_sources_worker(self) -> _DocsResult:
307 """Read the source count and the first page off the UI thread.
309 ``load_failed`` is True only when the store read actually raised;
310 an empty store with zero documents is the routine first-run state.
311 The read is one count and one page, so a library of millions of sources
312 costs the same screen as a library of ten.
313 """
314 try:
315 held_out, held_out_total = held_out_sources()
316 store = get_services().store
317 return _DocsResult(
318 sources=store.get_sources(limit=_DOCS_PAGE_SIZE, offset=0),
319 load_failed=False,
320 document_count=store.count_sources(),
321 held_out=held_out,
322 held_out_total=held_out_total,
323 )
324 except Exception:
325 log.debug("Failed to read store for status screen", exc_info=True)
326 return _DocsResult(sources=[], load_failed=True)
328 @work(thread=True, name=_WORKER_FETCH_DOCS_PAGE, exit_on_error=False)
329 def _fetch_docs_page_worker(self, offset: int) -> _DocsPage:
330 """Read one further page of sources off the UI thread."""
331 try:
332 page = get_services().store.get_sources(limit=_DOCS_PAGE_SIZE, offset=offset)
333 return _DocsPage(sources=page)
334 except Exception:
335 log.debug("Failed to read a page of sources for status screen", exc_info=True)
336 return _DocsPage(sources=[], load_failed=True)
338 @work(thread=True, name="status_fetch_arch", exit_on_error=False)
339 def _fetch_arch_worker(self) -> ModelArchInfo:
340 try:
341 return get_model_architecture()
342 except Exception:
343 log.debug("Failed to read model architecture for status", exc_info=True)
344 return ModelArchInfo()
346 def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
347 if event.state != WorkerState.SUCCESS:
348 return
349 if event.worker.name == _WORKER_FETCH_DOCS_PAGE:
350 page = event.worker.result
351 self._append_docs_page(page if isinstance(page, _DocsPage) else _DocsPage([], True))
352 elif event.worker.name == _WORKER_FETCH_SOURCES:
353 result = event.worker.result
354 docs = result if isinstance(result, _DocsResult) else _DocsResult([], True)
355 if self._sections_mounted:
356 self._apply_docs(docs)
357 else:
358 self._pending_docs = docs
359 elif event.worker.name == "status_fetch_arch":
360 arch = event.worker.result
361 if isinstance(arch, ModelArchInfo):
362 if self._sections_mounted:
363 self._load_arch(arch)
364 else:
365 self._pending_arch = arch
367 def _apply_docs(self, docs: _DocsResult) -> None:
368 """Render *docs* into the Documents table + storage section."""
369 self._load_documents(docs)
370 self._load_held_out(docs)
371 self._load_storage(docs.document_count)
373 def _load_held_out(self, docs: _DocsResult) -> None:
374 """Fill the held-out table; the section opens only when something is held out."""
375 from textual.css.query import NoMatches
377 with contextlib.suppress(NoMatches):
378 table = self.query_one("#held-out-table", DataTable)
379 table.clear(columns=True)
380 table.add_columns("File", "Reason")
381 if not docs.held_out:
382 table.add_row(msg.STATUS_HELD_OUT_EMPTY, "")
383 return
384 for held in docs.held_out:
385 table.add_row(Text(strip_control_characters(held.filename)), Text(held.reason))
386 hidden = docs.held_out_total - len(docs.held_out)
387 if hidden > 0:
388 table.add_row(msg.STATUS_HELD_OUT_MORE.format(count=hidden), "")
389 self.query_one("#held-out-section", Collapsible).collapsed = False
391 def _load_arch(self, info: ModelArchInfo) -> None:
392 """Populate the model architecture section from worker result."""
393 from textual.css.query import NoMatches
395 with contextlib.suppress(NoMatches):
396 self.query_one("#arch-info", Static).update(_build_arch_content(info))
398 def _load_config(self) -> None:
399 """Populate the configuration section."""
400 self.query_one("#config-info", Static).update(_build_config_content())
402 def _load_documents(self, docs: _DocsResult) -> None:
403 """Fill the table with the first page of sources and arm the paging."""
404 from textual.css.query import NoMatches
406 with contextlib.suppress(NoMatches):
407 table = self.query_one("#docs-table", DataTable)
408 table.clear()
409 self._docs_loaded = len(docs.sources)
410 self._docs_total = docs.document_count
411 self._docs_has_more = self._docs_loaded < self._docs_total
412 self._docs_loading = False
413 self._docs_page_failed = False
414 if not self._docs_scroll_watched:
415 self.watch(table, "scroll_y", self._on_docs_scrolled, init=False)
416 self._docs_scroll_watched = True
417 self._show_docs_hint()
418 if not docs.sources:
419 placeholder = (
420 msg.STATUS_DOCS_LOAD_FAILED if docs.load_failed else msg.STATUS_DOCS_EMPTY
421 )
422 table.add_row(placeholder, "")
423 return
424 for src in docs.sources:
425 table.add_row(
426 Text(strip_control_characters(src["filename"])), str(src["chunk_count"])
427 )
429 def _append_docs_page(self, page: _DocsPage) -> None:
430 """Add a further page of sources under the rows already shown."""
431 from textual.css.query import NoMatches
433 self._docs_loading = False
434 with contextlib.suppress(NoMatches):
435 table = self.query_one("#docs-table", DataTable)
436 for src in page.sources:
437 table.add_row(
438 Text(strip_control_characters(src["filename"])), str(src["chunk_count"])
439 )
440 self._docs_loaded += len(page.sources)
441 self._docs_page_failed = page.load_failed
442 # An empty page means the library shrank under the count; stop asking.
443 self._docs_has_more = bool(page.sources) and self._docs_loaded < self._docs_total
444 self._show_docs_hint()
446 def _show_docs_hint(self) -> None:
447 """Say how many sources are loaded while more remain, and that a read failed."""
448 hint = self.query_one("#docs-more", Static)
449 if self._docs_page_failed:
450 hint.update(msg.STATUS_DOCS_LOAD_FAILED)
451 elif self._docs_has_more:
452 hint.update(msg.STATUS_DOCS_MORE.format(count=self._docs_loaded))
453 hint.display = self._docs_page_failed or self._docs_has_more
455 def _load_more_docs(self) -> None:
456 """Start the next page unless one is already loading or none is left."""
457 if self._docs_loading or not self._docs_has_more:
458 return
459 self._docs_loading = True
460 self._fetch_docs_page_worker(self._docs_loaded)
462 def _on_docs_scrolled(self, _scroll_y: float) -> None:
463 """Load the next page when the user scrolls near the bottom of the table."""
464 table = self.query_one("#docs-table", DataTable)
465 max_y = table.max_scroll_y
466 if max_y > 0 and table.scroll_y / max_y >= _DOCS_SCROLL_PREFETCH_RATIO:
467 self._load_more_docs()
469 def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
470 """Load the next page when the cursor nears the last loaded source."""
471 near_end = event.cursor_row >= self._docs_loaded * _DOCS_SCROLL_PREFETCH_RATIO
472 if event.data_table.id == "docs-table" and near_end:
473 self._load_more_docs()
475 def _load_storage(self, doc_count: int) -> None:
476 """Populate the storage section."""
477 from textual.css.query import NoMatches
479 with contextlib.suppress(NoMatches):
480 self.query_one("#storage-info", Static).update(_build_storage_content(doc_count))
482 def action_go_back(self) -> None:
483 self.app.go_back()
485 def action_cursor_down(self) -> None:
486 self.query_one("#status-scroll", VerticalScroll).scroll_down()
488 def action_cursor_up(self) -> None:
489 self.query_one("#status-scroll", VerticalScroll).scroll_up()
491 def action_jump_top(self) -> None:
492 self.query_one("#status-scroll", VerticalScroll).scroll_home(animate=False)
494 def action_jump_bottom(self) -> None:
495 self.query_one("#status-scroll", VerticalScroll).scroll_end(animate=False)