Coverage for src/lilbee/cli/tui/screens/startup_gate.py: 100%
111 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Startup splash: the lilbee wordmark while the services build and the library opens."""
3from __future__ import annotations
5import logging
6from collections.abc import Callable
7from time import monotonic
9from textual import events, work
10from textual.app import ComposeResult
11from textual.containers import Vertical
12from textual.screen import Screen
13from textual.timer import Timer
14from textual.widgets import ProgressBar, Static
15from textual.worker import get_current_worker
17from lilbee.app.services import get_services, peek_services, report_library_upgrade
18from lilbee.cli.tui import messages as msg
19from lilbee.cli.tui.app import LilbeeApp
20from lilbee.runtime.bee_logo import BEE_LINES
21from lilbee.runtime.progress import (
22 UPGRADE_HEADLINE,
23 EventType,
24 LibraryUpgradeEvent,
25 ProgressEvent,
26 UpgradePhase,
27)
29log = logging.getLogger(__name__)
31_LOGO = "\n".join(BEE_LINES)
32# How often the status line redraws the elapsed time of a library upgrade.
33_UPGRADE_TICK_SECONDS = 1.0
34# How long the finished upgrade's toast stays on the landing view.
35_UPGRADE_DONE_TOAST_SECONDS = 6
38class StartupGate(Screen[None]):
39 """Holds the screen until readiness settles, then hands over to the landing view.
41 The engine itself loads in the background after the handover; a prompt sent
42 before it is ready waits inside its own answer bubble with live progress.
43 """
45 CSS_PATH = "startup_gate.tcss"
47 # Lilbee always hosts screens on a LilbeeApp, so narrowing the type lets
48 # reveal_landing resolve without reflection.
49 app: LilbeeApp # type: ignore[assignment]
51 # The app reference every worker and UI hop uses: ``self.app`` reads a
52 # contextvar that plain threads never carry and then walks ``_parent``,
53 # which raises NoActiveAppError the moment the gate detaches. Captured
54 # once in on_mount, on the UI thread, where resolution is certain.
55 _ui_app: LilbeeApp
57 def __init__(self) -> None:
58 super().__init__()
59 # The latest event of a library upgrade that runs, when it began, and the
60 # timer that redraws its time; no timer while none runs.
61 self._upgrade_event = LibraryUpgradeEvent(phase=UpgradePhase.WAITING)
62 self._upgrade_started = 0.0
63 self._upgrade_timer: Timer | None = None
64 # Printable keys pressed while the gate shows; the prompt receives them.
65 self._typed: list[str] = []
67 def on_key(self, event: events.Key) -> None:
68 """Keep printable keys for the prompt so no view binding takes them."""
69 if event.is_printable and event.character:
70 self._typed.append(event.character)
71 event.prevent_default()
72 event.stop()
74 def on_paste(self, event: events.Paste) -> None:
75 """Keep pasted text for the prompt, as for typed keys."""
76 self._typed.append(event.text)
77 event.prevent_default()
78 event.stop()
80 def compose(self) -> ComposeResult:
81 with Vertical(id="gate-body"):
82 yield Static(_LOGO, id="gate-logo")
83 yield ProgressBar(total=None, show_eta=False, show_percentage=False, id="gate-bar")
84 yield Static(msg.STARTUP_PREPARING, id="gate-status")
86 def on_mount(self) -> None:
87 """Capture the app for the workers, then retire the launcher's splash.
89 The splash animates over the blank alt-screen right up to this moment,
90 so the wordmark never leaves the terminal. Dismissal waits on the
91 subprocess, so it runs off-thread; the refresh afterwards repaints
92 anything a final splash frame may have touched.
93 """
94 self._ui_app = self.app
95 self._retire_splash()
97 @work(thread=True, name="splash_retire", exit_on_error=False)
98 def _retire_splash(self) -> None:
99 from lilbee.runtime.splash import dismiss
101 dismiss()
102 self._marshal(self._repaint)
104 def _repaint(self) -> None:
105 """Repaint anything a final splash frame may have scribbled over."""
106 self.refresh()
108 def start_boot(self) -> None:
109 """Work out what the app can serve, off the UI thread, then hand over.
111 One path even when the container is already built (a second TUI in the
112 same process, a test host): the setup answer still decides whether the
113 handover happens. The thread hop also keeps the screen switch out of
114 on_mount, which stalls Textual.
115 """
116 self._boot_worker()
118 @work(thread=True, name="startup_gate", exit_on_error=False)
119 def _boot_worker(self) -> None:
120 """Settle the refs, settle readiness, build the container, hand over.
122 Canonicalization runs first: it can swap a stale ref for a working one,
123 or adopt an installed model into an unconfigured role, which decides
124 the landing view. An already built container skips it.
126 ``settle_landing`` blocks until the app has recorded the answer, so
127 the handover cannot read a readiness flag that has yet to be written.
129 Building the container spawns the role servers, so it belongs on this
130 thread, behind the loading bar, and so does opening the library: an open
131 that upgrades it shows each step in the status line.
132 """
133 report_library_upgrade(self._show_upgrade)
134 try:
135 if peek_services() is None:
136 self._ui_app.canonicalize_persisted_models()
137 self._ui_app.settle_landing()
138 self._ui_app.adopt_services()
139 get_services().store.get_db()
140 except Exception as exc:
141 # Any failure to prepare the app leaves the user with no engine.
142 # Show it and hand them the rest of the TUI rather than a dead screen.
143 log.exception("startup gate could not prepare the app")
144 self._marshal(self._fail, str(exc))
145 return
146 self._marshal(self._release)
148 def _show_upgrade(self, _event_type: EventType, event: ProgressEvent) -> None:
149 """Hand a library upgrade event to the UI thread, from the opening thread."""
150 # The reporter takes the whole progress union; the store passes it upgrade events only.
151 if isinstance(event, LibraryUpgradeEvent) and self.is_mounted:
152 self._ui_app.call_from_thread(self._show_event, event)
154 def _show_event(self, event: LibraryUpgradeEvent) -> None:
155 """Show *event* under the bar; while the upgrade runs its time redraws each second.
157 The finished line is also a toast, so it outlives the gate's handover.
158 """
159 if not event.running:
160 self._stop_ticking()
161 self._set_status(event.text)
162 if event.phase is UpgradePhase.DONE:
163 self._ui_app.notify(event.text, timeout=_UPGRADE_DONE_TOAST_SECONDS)
164 return
165 self._upgrade_event = event
166 self._upgrade_started = monotonic() - event.elapsed_s
167 if self._upgrade_timer is None:
168 self._upgrade_timer = self.set_interval(_UPGRADE_TICK_SECONDS, self._tick_upgrade)
169 self._tick_upgrade()
171 def _tick_upgrade(self) -> None:
172 """Redraw the upgrade's headline and its progress line, timed to now."""
173 line = self._upgrade_event.line_at(monotonic() - self._upgrade_started)
174 self._set_status(f"{UPGRADE_HEADLINE}\n{line}")
176 def _stop_ticking(self) -> None:
177 """Stop redrawing the upgrade's time; safe when it never started."""
178 if self._upgrade_timer is not None:
179 self._upgrade_timer.stop()
180 self._upgrade_timer = None
182 def _set_status(self, text: str) -> None:
183 """Put *text* in the status line under the bar."""
184 for status in self.query("#gate-status").results(Static):
185 status.update(text)
187 def _stopping(self) -> bool:
188 """True once the worker was cancelled or the gate left the screen."""
189 worker = get_current_worker()
190 return worker.is_cancelled or not self.is_mounted
192 def _marshal(self, callback: Callable[..., None], *args: object) -> None:
193 """Hop to the UI thread, unless the app is already tearing down."""
194 if self._stopping():
195 return
196 self._ui_app.call_from_thread(callback, *args)
198 def _fail(self, error: str) -> None:
199 """Surface a failed start and hand the user to the rest of the TUI to fix it."""
200 self._stop_ticking()
201 self._ui_app.notify(msg.STARTUP_FAILED.format(error=error), severity="error", timeout=8)
202 self._release()
204 def _release(self) -> None:
205 """Hand the screen to the landing view, unless something else has taken it.
207 reveal_landing switches whatever screen is on top, so a gate that resolved
208 after another screen opened above it would replace that screen instead of
209 itself. No widget lookup here either: the gate can resolve before compose
210 has mounted its children, and a missed query would strand the user.
211 """
212 if self._ui_app.screen is not self:
213 return
214 self._ui_app.type_ahead = "".join(self._typed)
215 self._ui_app.reveal_landing()