Coverage for src/lilbee/cli/tui/screens/startup_gate.py: 100%

1"""Startup splash: the lilbee wordmark while the services container builds.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from collections.abc import Callable 

7 

8from textual import work 

9from textual.app import ComposeResult 

10from textual.containers import Vertical 

11from textual.screen import Screen 

12from textual.widgets import ProgressBar, Static 

13from textual.worker import get_current_worker 

14 

15from lilbee.app.services import peek_services 

16from lilbee.cli.tui import messages as msg 

17from lilbee.cli.tui.app import LilbeeApp 

18from lilbee.runtime.bee_logo import BEE_LINES 

19 

20log = logging.getLogger(__name__) 

21 

22_LOGO = "\n".join(BEE_LINES) 

23 

24 

25class StartupGate(Screen[None]): 

26 """Holds the screen until readiness settles, then hands over to the landing view. 

27 

28 The engine itself loads in the background after the handover; a prompt sent 

29 before it is ready waits inside its own answer bubble with live progress. 

30 """ 

31 

32 CSS_PATH = "startup_gate.tcss" 

33 

34 # Lilbee always hosts screens on a LilbeeApp, so narrowing the type lets 

35 # reveal_landing resolve without reflection. 

36 app: LilbeeApp # type: ignore[assignment] 

37 

38 # The app reference every worker and UI hop uses: ``self.app`` reads a 

39 # contextvar that plain threads never carry and then walks ``_parent``, 

40 # which raises NoActiveAppError the moment the gate detaches. Captured 

41 # once in on_mount, on the UI thread, where resolution is certain. 

42 _ui_app: LilbeeApp 

43 

44 def compose(self) -> ComposeResult: 

45 with Vertical(id="gate-body"): 

46 yield Static(_LOGO, id="gate-logo") 

47 yield ProgressBar(total=None, show_eta=False, show_percentage=False, id="gate-bar") 

48 yield Static(msg.STARTUP_PREPARING, id="gate-status") 

49 

50 def on_mount(self) -> None: 

51 """Capture the app for the workers, then retire the launcher's splash. 

52 

53 The splash animates over the blank alt-screen right up to this moment, 

54 so the wordmark never leaves the terminal. Dismissal waits on the 

55 subprocess, so it runs off-thread; the refresh afterwards repaints 

56 anything a final splash frame may have touched. 

57 """ 

58 self._ui_app = self.app 

59 self._retire_splash() 

60 

61 @work(thread=True, name="splash_retire", exit_on_error=False) 

62 def _retire_splash(self) -> None: 

63 from lilbee.runtime.splash import dismiss 

64 

65 dismiss() 

66 self._marshal(self._repaint) 

67 

68 def _repaint(self) -> None: 

69 """Repaint anything a final splash frame may have scribbled over.""" 

70 self.refresh() 

71 

72 def start_boot(self) -> None: 

73 """Work out what the app can serve, off the UI thread, then hand over. 

74 

75 One path even when the container is already built (a second TUI in the 

76 same process, a test host): the setup answer still decides whether the 

77 handover happens. The thread hop also keeps the screen switch out of 

78 on_mount, which stalls Textual. 

79 """ 

80 self._boot_worker() 

81 

82 @work(thread=True, name="startup_gate", exit_on_error=False) 

83 def _boot_worker(self) -> None: 

84 """Settle the refs, settle readiness, build the container, hand over. 

85 

86 Canonicalization runs first: it can swap a stale ref for a working one, 

87 or adopt an installed model into an unconfigured role, which decides 

88 the landing view. An already built container skips it. 

89 

90 ``settle_landing`` blocks until the app has recorded the answer, so 

91 the handover cannot read a readiness flag that has yet to be written. 

92 

93 Building the container spawns the role servers, so it belongs on this 

94 thread, behind the loading bar. 

95 """ 

96 try: 

97 if peek_services() is None: 

98 self._ui_app.canonicalize_persisted_models() 

99 self._ui_app.settle_landing() 

100 self._ui_app.adopt_services() 

101 except Exception as exc: 

102 # Any failure to prepare the app leaves the user with no engine. 

103 # Show it and hand them the rest of the TUI rather than a dead screen. 

104 log.exception("startup gate could not prepare the app") 

105 self._marshal(self._fail, str(exc)) 

106 return 

107 self._marshal(self._release) 

108 

109 def _stopping(self) -> bool: 

110 """True once the worker was cancelled or the gate left the screen.""" 

111 worker = get_current_worker() 

112 return worker.is_cancelled or not self.is_mounted 

113 

114 def _marshal(self, callback: Callable[..., None], *args: object) -> None: 

115 """Hop to the UI thread, unless the app is already tearing down.""" 

116 if self._stopping(): 

117 return 

118 self._ui_app.call_from_thread(callback, *args) 

119 

120 def _fail(self, error: str) -> None: 

121 """Surface a failed start and hand the user to the rest of the TUI to fix it.""" 

122 self._ui_app.notify(msg.STARTUP_FAILED.format(error=error), severity="error", timeout=8) 

123 self._release() 

124 

125 def _release(self) -> None: 

126 """Hand the screen to the landing view, unless something else has taken it. 

127 

128 reveal_landing switches whatever screen is on top, so a gate that resolved 

129 after another screen opened above it would replace that screen instead of 

130 itself. No widget lookup here either: the gate can resolve before compose 

131 has mounted its children, and a missed query would strand the user. 

132 """ 

133 if self._ui_app.screen is not self: 

134 return 

135 self._ui_app.reveal_landing()