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

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

2 

3from __future__ import annotations 

4 

5import logging 

6from typing import TYPE_CHECKING, Any, cast 

7 

8from textual.command import Hit, Hits, Provider 

9 

10from lilbee.app.services import get_services 

11from lilbee.cli.tui import messages as msg 

12from lilbee.core.config import cfg 

13 

14log = logging.getLogger(__name__) 

15 

16if TYPE_CHECKING: 

17 from lilbee.cli.tui.app import LilbeeApp 

18 

19 

20class LilbeeCommandProvider(Provider): 

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

22 

23 @property 

24 def _app(self) -> LilbeeApp: 

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

26 

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

28 matcher = self.matcher(query) 

29 for cmd_text, help_text, action in self._get_commands(): 

30 score = matcher.match(cmd_text) 

31 if score > 0: 

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

33 

34 async def discover(self) -> Hits: 

35 for cmd_text, help_text, action in self._get_commands(): 

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

37 

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

39 app = self._app 

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

41 ("Open catalog", "Browse and install models", lambda: app.switch_view("Catalog")), 

42 ("Run setup wizard", "Configure chat and embedding models", self._action_setup), 

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

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

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

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

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

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

49 ( 

50 "Retry skipped documents", 

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

52 self._action_retry_skipped, 

53 ), 

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

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

56 ( 

57 "Reset knowledge base", 

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

59 self._action_reset, 

60 ), 

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

62 ] 

63 

64 commands.extend(self._model_commands()) 

65 commands.extend(self._document_commands()) 

66 return commands 

67 

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

69 """Generate commands for installed models.""" 

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

71 try: 

72 from lilbee.modelhub.models import list_installed_models 

73 

74 for name in list_installed_models(): 

75 commands.append( 

76 ( 

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

78 "Switch chat model", 

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

80 ) 

81 ) 

82 except Exception: 

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

84 

85 return commands 

86 

87 def _document_commands(self) -> list[tuple[str, str, Any]]: 

88 """Generate commands for indexed documents.""" 

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

90 try: 

91 for src in get_services().store.get_sources(): 

92 name = src.get("filename", src.get("source", "")) 

93 if name: 

94 commands.append( 

95 ( 

96 f"Delete document → {name}", 

97 f"Remove {name} from index", 

98 lambda n=name: self._delete_doc(n), 

99 ) 

100 ) 

101 except Exception: 

102 log.debug("Failed to list documents", exc_info=True) 

103 return commands 

104 

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

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

107 # / status bar subscribers (settings_changed_signal) refresh. 

108 app = self._app 

109 app.set_active_model(attr, value) 

110 display = value or "off" 

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

112 if attr == "chat_model": 

113 app.title = f"lilbee: {value}" 

114 

115 def _delete_doc(self, name: str) -> None: 

116 get_services().store.remove_documents([name]) 

117 self.screen.app.notify(f"Deleted {name}") 

118 

119 def _action_sync(self) -> None: 

120 self._app.action_run_sync() 

121 

122 def _action_retry_skipped(self) -> None: 

123 """Clear the failed-file markers and kick off a sync to retry them. 

124 

125 Clearing the marker cache and then running a normal sync is 

126 equivalent to ``lilbee sync --retry-skipped`` / ``POST /api/sync`` 

127 with ``retry_skipped=true``. 

128 """ 

129 from lilbee.data.ingest.skip_marker import clear_skip_markers, load_skip_markers 

130 

131 cleared = len(load_skip_markers(cfg.data_root)) 

132 clear_skip_markers(cfg.data_root) 

133 self.screen.app.notify(msg.retry_skipped_message(cleared)) 

134 self._app.action_run_sync() 

135 

136 def _action_version(self) -> None: 

137 from lilbee.app.version import get_version 

138 

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

140 

141 def _action_setup(self) -> None: 

142 from lilbee.cli.tui.screens.setup import SetupWizard 

143 

144 self.screen.app.push_screen(SetupWizard()) 

145 

146 def _action_open_wiki(self) -> None: 

147 self._app.switch_view("Wiki") 

148 

149 def _action_reset(self) -> None: 

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

151 from lilbee.cli.tui.screens.chat import ChatScreen 

152 

153 app = self._app 

154 chat = next((s for s in app.screen_stack if isinstance(s, ChatScreen)), None) 

155 if chat is None: 

156 app.notify("Open Chat to run /reset") 

157 return 

158 chat._cmd_reset("")