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

1"""Command palette provider for lilbee TUI.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import logging 

7from typing import TYPE_CHECKING, Any, cast 

8 

9from textual.command import CommandInput, Hit, Hits, Provider 

10from textual.widgets import Input 

11 

12from lilbee.app.services import get_services 

13from lilbee.catalog import display_label_for_ref 

14from lilbee.cli.tui import messages as msg 

15from lilbee.cli.tui.command_registry import COMMANDS, SlashCommand, get_command 

16from lilbee.cli.tui.thread_safe import call_from_thread 

17from lilbee.runtime.lock import LockTimeoutError 

18 

19log = logging.getLogger(__name__) 

20 

21if TYPE_CHECKING: 

22 from textual.screen import Screen 

23 from textual.style import Style 

24 

25 from lilbee.cli.tui.app import LilbeeApp 

26 

27 

28class LilbeeCommandProvider(Provider): 

29 """Provides searchable commands for the Textual command palette (Ctrl+P).""" 

30 

31 def __init__(self, screen: Screen[Any], match_style: Style | None = None) -> None: 

32 super().__init__(screen, match_style) 

33 self._models: list[tuple[str, str, Any]] = [] 

34 self._model_lookup: asyncio.Task[list[tuple[str, str, Any]]] | None = None 

35 

36 @property 

37 def _app(self) -> LilbeeApp: 

38 return cast("LilbeeApp", self.screen.app) 

39 

40 async def startup(self) -> None: 

41 """Start the model lookup on a thread, once per palette open. 

42 

43 The lookup asks each local server over HTTP, which can take seconds. 

44 No search waits for it: the static commands match at once, and the 

45 model rows join when the lookup lands. 

46 """ 

47 palette = self.app.screen 

48 self._model_lookup = asyncio.create_task(asyncio.to_thread(self._model_commands)) 

49 self._model_lookup.add_done_callback(lambda done: self._models_landed(palette, done)) 

50 

51 def _models_landed( 

52 self, palette: Screen[Any], lookup: asyncio.Future[list[tuple[str, str, Any]]] 

53 ) -> None: 

54 """Keep the model rows and search again, unless the palette has closed. 

55 

56 The search runs again only when a model row matches what the user typed, 

57 because a fresh search moves the highlight back to the first row. 

58 """ 

59 if lookup.cancelled(): 

60 return 

61 self._models = lookup.result() 

62 if self.app.screen is not palette: 

63 return 

64 # An app that exits removes the palette's input while the palette is still on top. 

65 for command_input in palette.query(CommandInput): 

66 query = command_input.value.strip() 

67 matcher = self.matcher(query) 

68 if any(not query or matcher.match(text) > 0 for text, _, _ in self._models): 

69 palette.post_message(Input.Changed(command_input, command_input.value)) 

70 

71 async def search(self, query: str) -> Hits: 

72 matcher = self.matcher(query) 

73 for cmd_text, help_text, action in self._all_commands(): 

74 score = matcher.match(cmd_text) 

75 if score > 0: 

76 yield Hit(score, matcher.highlight(cmd_text), action, help=help_text) 

77 

78 async def discover(self) -> Hits: 

79 for cmd_text, help_text, action in self._all_commands(): 

80 yield Hit(1.0, cmd_text, action, help=help_text) 

81 

82 def _all_commands(self) -> list[tuple[str, str, Any]]: 

83 return [*self._get_commands(), *self._models] 

84 

85 def _get_commands(self) -> list[tuple[str, str, Any]]: 

86 app = self._app 

87 commands: list[tuple[str, str, Any]] = [ 

88 ("Open chat", "Ask questions about your knowledge base", app.action_open_chat), 

89 ("Open catalog", "Browse and install models", app.action_open_catalog), 

90 ("Open status", "Knowledge base status", lambda: app.switch_view("Status")), 

91 ("Open settings", "View and change settings", lambda: app.switch_view("Settings")), 

92 ("Choose profile", "Pick the settings profile for this project", app.open_profile_tab), 

93 ("Analyze documents", "Recommend a profile for your documents", app.start_analyze), 

94 ("Open task center", "Monitor background tasks", lambda: app.switch_view("Tasks")), 

95 ("Help", "Show keybinding reference", app.action_push_help), 

96 ("Cycle theme", "Switch to next color theme", app.action_cycle_theme), 

97 ("Sync documents", "Sync knowledge base", self._action_sync), 

98 ( 

99 "Retry skipped documents", 

100 "Re-attempt files that failed a previous sync", 

101 self._action_retry_skipped, 

102 ), 

103 ( 

104 "Delete document", 

105 "Remove a file from the index (Tab completes names)", 

106 self._action_delete_document, 

107 ), 

108 ( 

109 "Prune ignored documents", 

110 "Drop indexed documents a .lilbeeignore now excludes", 

111 self._action_prune_ignored, 

112 ), 

113 ("Open wiki", "Browse and generate wiki pages", self._action_open_wiki), 

114 ( 

115 "Wikify", 

116 "Generate wiki pages from indexed documents (GPU-heavy)", 

117 self._action_wikify, 

118 ), 

119 ( 

120 "Delete wiki", 

121 "Remove every generated wiki page and its indexed rows", 

122 self._action_wipe_wiki, 

123 ), 

124 ("Show version", "Display lilbee version", self._action_version), 

125 ( 

126 "Reset knowledge base", 

127 "Delete all data (asks for confirmation)", 

128 self._action_reset, 

129 ), 

130 ("Quit", "Exit lilbee", app.action_quit), 

131 ] 

132 

133 commands.extend(self._slash_commands()) 

134 return commands 

135 

136 def _slash_commands(self) -> list[tuple[str, str, Any]]: 

137 """One palette entry per slash command, mirroring the chat surface.""" 

138 return [ 

139 (cmd.name, cmd.help_text, lambda c=cmd: self._run_slash_command(c)) for cmd in COMMANDS 

140 ] 

141 

142 def _run_slash_command(self, cmd: SlashCommand) -> None: 

143 """Run *cmd* through Chat: dispatch it, or prefill it when it needs arguments.""" 

144 app = self._app 

145 chat = app.chat_screen() 

146 if chat is None: 

147 app.notify(f"Open Chat to run {cmd.name}") 

148 return 

149 if cmd.args_hint.startswith("<"): 

150 # Needs an argument: land in the chat prompt for Tab completion. 

151 app.switch_view(msg.DEFAULT_VIEW) 

152 chat.insert_slash_command(cmd.name) 

153 else: 

154 # Complete as-is: dispatch like a submitted prompt. Handlers that 

155 # navigate call switch_view themselves, and switch_view no-ops 

156 # while another switch is in flight, so don't pre-switch to Chat. 

157 chat.run_command(cmd.name) 

158 

159 def _model_commands(self) -> list[tuple[str, str, Any]]: 

160 """Generate commands for installed models.""" 

161 commands: list[tuple[str, str, Any]] = [] 

162 try: 

163 from lilbee.modelhub.models import list_installed_models 

164 

165 for name in list_installed_models(): 

166 commands.append( 

167 ( 

168 f"Set chat model → {name}", 

169 "Switch chat model", 

170 lambda n=name: self._set_model("chat_model", n), 

171 ) 

172 ) 

173 except Exception: 

174 log.debug("Failed to list installed models", exc_info=True) 

175 

176 return commands 

177 

178 def _set_model(self, attr: str, value: str) -> None: 

179 # Route through LilbeeApp.set_active_model so model-bar / scope chip 

180 # / status bar subscribers (settings_changed_signal) refresh. 

181 app = self._app 

182 app.set_active_model(attr, value) 

183 display = display_label_for_ref(value) or "off" 

184 app.notify(f"{attr}: {display}") 

185 if attr == "chat_model": 

186 app.title = msg.app_title(value) 

187 

188 def _action_delete_document(self) -> None: 

189 """Jump to Chat with /delete prefilled; Tab there completes file names.""" 

190 self._run_slash_command(get_command("/delete")) 

191 

192 def _action_sync(self) -> None: 

193 self._app.action_run_sync() 

194 

195 def _action_retry_skipped(self) -> None: 

196 """Release the failed files and start a sync, like ``lilbee sync --retry-skipped``.""" 

197 self._app.run_worker(self._retry_skipped, thread=True, exit_on_error=False) 

198 

199 def _retry_skipped(self) -> None: 

200 """Release the failed files off the UI thread, since the index write lock can wait.""" 

201 app = self._app 

202 try: 

203 cleared = get_services().store.release_failed() 

204 except LockTimeoutError as exc: 

205 call_from_thread(app, app.notify, str(exc), severity="error") 

206 return 

207 call_from_thread(app, app.notify, msg.retry_skipped_message(len(cleared))) 

208 call_from_thread(app, app.action_run_sync) 

209 

210 def _action_prune_ignored(self) -> None: 

211 """Sync with pruning on, the TUI equivalent of ``lilbee sync --prune-ignored``.""" 

212 self._run_slash_command(get_command("/prune-ignored")) 

213 

214 def _action_version(self) -> None: 

215 from lilbee.app.version import get_version 

216 

217 self.screen.app.notify(f"lilbee {get_version()}") 

218 

219 def _action_open_wiki(self) -> None: 

220 self._app.switch_view("Wiki") 

221 

222 def _action_wikify(self) -> None: 

223 from lilbee.cli.tui.screens.wiki import start_wikify 

224 

225 start_wikify(self._app) 

226 

227 def _action_wipe_wiki(self) -> None: 

228 """Delete the generated wiki. 

229 

230 The palette is the only TUI route to this while the wiki is off, since 

231 the wiki view (and its ``W`` binding) is dropped from the nav in that 

232 state, which is exactly when the leftover pages need removing. 

233 """ 

234 from lilbee.cli.tui.screens.wiki import confirm_wiki_wipe 

235 

236 confirm_wiki_wipe(self._app) 

237 

238 def _action_reset(self) -> None: 

239 """Trigger /reset from the palette so the ConfirmDialog flow fires.""" 

240 self._run_slash_command(get_command("/reset"))