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 import settings 

13from lilbee.core.config import cfg 

14 

15log = logging.getLogger(__name__) 

16 

17if TYPE_CHECKING: 

18 from lilbee.cli.tui.app import LilbeeApp 

19 

20 

21class LilbeeCommandProvider(Provider): 

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

23 

24 @property 

25 def _app(self) -> LilbeeApp: 

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

27 

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

29 matcher = self.matcher(query) 

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

31 score = matcher.match(cmd_text) 

32 if score > 0: 

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

34 

35 async def discover(self) -> Hits: 

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

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

38 

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

40 app = self._app 

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

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

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

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

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

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

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

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

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

50 ( 

51 "Retry skipped documents", 

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

53 self._action_retry_skipped, 

54 ), 

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

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

57 ( 

58 "Reset knowledge base", 

59 "Delete all data (requires /reset confirm)", 

60 self._action_noop, 

61 ), 

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

63 ] 

64 

65 commands.extend(self._model_commands()) 

66 commands.extend(self._document_commands()) 

67 return commands 

68 

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

70 """Generate commands for installed models.""" 

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

72 try: 

73 from lilbee.modelhub.models import list_installed_models 

74 

75 for name in list_installed_models(): 

76 commands.append( 

77 ( 

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

79 "Switch chat model", 

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

81 ) 

82 ) 

83 except Exception: 

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

85 

86 return commands 

87 

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

89 """Generate commands for indexed documents.""" 

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

91 try: 

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

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

94 if name: 

95 commands.append( 

96 ( 

97 f"Delete document → {name}", 

98 f"Remove {name} from index", 

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

100 ) 

101 ) 

102 except Exception: 

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

104 return commands 

105 

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

107 setattr(cfg, attr, value) 

108 settings.set_value(cfg.data_root, attr, value) 

109 display = value or "off" 

110 self.screen.app.notify(f"{attr}: {display}") 

111 if attr == "chat_model": 

112 self.screen.app.title = f"lilbee: {value}" 

113 

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

115 store = get_services().store 

116 store.delete_by_source(name) 

117 store.delete_source(name) 

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

119 

120 def _action_sync(self) -> None: 

121 self._app.action_run_sync() 

122 

123 def _action_retry_skipped(self) -> None: 

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

125 

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

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

128 with ``retry_skipped=true``. 

129 """ 

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

131 

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

133 clear_skip_markers(cfg.data_root) 

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

135 self._app.action_run_sync() 

136 

137 def _action_version(self) -> None: 

138 from lilbee.app.version import get_version 

139 

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

141 

142 def _action_setup(self) -> None: 

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

144 

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

146 

147 def _action_open_wiki(self) -> None: 

148 self._app.switch_view("Wiki") 

149 

150 def _action_noop(self) -> None: 

151 self.screen.app.notify("Type '/reset confirm' in chat to reset")