Coverage for src/lilbee/cli/model.py: 100%

1"""`lilbee model` sub-app: list/show/pull/rm/browse for installed models. 

2 

3Thin Typer wrapper around the surface-agnostic use-cases in 

4:mod:`lilbee.app.models`. Bare result models live in ``app.models``; 

5the Rich renderers below adapt them for human-readable terminal output. 

6""" 

7 

8from __future__ import annotations 

9 

10import sys 

11from pathlib import Path 

12from typing import TYPE_CHECKING 

13 

14import typer 

15from rich.console import Console 

16from rich.progress import BarColumn, Progress, TextColumn, TimeRemainingColumn 

17from rich.table import Table 

18 

19from lilbee.app.models import ( 

20 ListModelsResult, 

21 PullEvent, 

22 PullProgressEvent, 

23 PullResult, 

24 PullStatus, 

25 ShowModelResult, 

26 list_models_data, 

27 pull_model_data, 

28 remove_model_data, 

29 show_model_data, 

30) 

31from lilbee.cli import theme 

32from lilbee.cli.app import ( 

33 apply_overrides, 

34 console, 

35 data_dir_option, 

36 global_option, 

37) 

38from lilbee.cli.helpers import json_output 

39from lilbee.core.config import cfg 

40 

41if TYPE_CHECKING: 

42 from collections.abc import Callable 

43 

44 from lilbee.catalog import DownloadProgress 

45 from lilbee.catalog.types import ModelSource, ModelTask 

46 

47 

48def _render_list(data: ListModelsResult) -> Table: 

49 table = Table(title="Installed models") 

50 table.add_column("Name", style=theme.ACCENT) 

51 table.add_column("Source", style=theme.MUTED) 

52 table.add_column("Task") 

53 table.add_column("Size", justify="right") 

54 for entry in data.models: 

55 size = f"{entry.size_gb:.2f} GB" if entry.size_gb is not None else "" 

56 table.add_row(entry.name, entry.source, entry.task or "", size) 

57 return table 

58 

59 

60def _render_show(data: ShowModelResult) -> str: 

61 lines = [f"[{theme.ACCENT}]{data.model}[/{theme.ACCENT}]"] 

62 if data.catalog is not None: 

63 lines.extend( 

64 [ 

65 f" display_name: {data.catalog.display_name}", 

66 f" task: {data.catalog.task}", 

67 f" size_gb: {data.catalog.size_gb}", 

68 f" min_ram_gb: {data.catalog.min_ram_gb}", 

69 f" hf_repo: {data.catalog.hf_repo}", 

70 f" description: {data.catalog.description}", 

71 ] 

72 ) 

73 lines.append(f" installed: {data.installed}") 

74 if data.source: 

75 lines.append(f" source: {data.source}") 

76 if data.path: 

77 lines.append(f" path: {data.path}") 

78 if data.manifest is not None: 

79 lines.append(f" downloaded: {data.manifest.downloaded_at}") 

80 return "\n".join(lines) 

81 

82 

83model_app = typer.Typer( 

84 name="model", 

85 help="Manage installed and available models (pull / list / show / rm / browse).", 

86 no_args_is_help=True, 

87) 

88 

89_source_option = typer.Option( 

90 None, 

91 "--source", 

92 "-s", 

93 help="Filter by source: native, remote, ollama, lm_studio, or frontier (default: all).", 

94) 

95_task_option = typer.Option( 

96 None, 

97 "--task", 

98 "-t", 

99 help="Filter by task: 'chat', 'embedding', 'vision', or 'rerank'.", 

100) 

101_yes_option = typer.Option( 

102 False, 

103 "--yes", 

104 "-y", 

105 help="Skip confirmation prompt.", 

106) 

107 

108 

109def _parse_source_or_bad_param(value: str | None) -> ModelSource | None: 

110 """Parse a CLI --source value, raising typer.BadParameter on bad input.""" 

111 from lilbee.catalog.types import ModelSource 

112 

113 try: 

114 return ModelSource.parse(value) 

115 except ValueError as exc: 

116 if cfg.json_mode: 

117 json_output({"error": str(exc)}) 

118 raise SystemExit(1) from None 

119 raise typer.BadParameter(str(exc)) from exc 

120 

121 

122def _parse_task_or_bad_param(value: str | None) -> ModelTask | None: 

123 """Parse a CLI --task value, mirroring --source's JSON-envelope + friendly error. 

124 

125 Without this, an invalid --task leaked Python's "'x' is not a valid ModelTask" 

126 and broke --json by emitting Typer usage text instead of the error envelope. 

127 """ 

128 from lilbee.catalog.types import ModelTask 

129 

130 if not value: 

131 return None 

132 try: 

133 return ModelTask(value) 

134 except ValueError: 

135 allowed = ", ".join(t.value for t in ModelTask) 

136 msg = f"invalid task {value!r}; expected one of: {allowed}" 

137 if cfg.json_mode: 

138 json_output({"error": msg}) 

139 raise SystemExit(1) from None 

140 raise typer.BadParameter(msg) from None 

141 

142 

143def _apply_catalog_overrides(*, data_dir: Path | None, use_global: bool) -> None: 

144 """Apply CLI overrides for a catalog command, with the fleet left cold. 

145 

146 Catalog commands list, read, download and delete model files; none of them 

147 runs inference, so none should pay for a fleet warm (or its alarming 

148 warm-up traceback on a slow host). Setting it here rather than at each 

149 command means a catalog command added later cannot silently eager-start. 

150 """ 

151 apply_overrides(data_dir=data_dir, use_global=use_global) 

152 cfg.worker_pool_eager_start = False 

153 

154 

155@model_app.command("list") 

156def list_cmd( 

157 source: str | None = _source_option, 

158 task: str | None = _task_option, 

159 data_dir: Path | None = data_dir_option, 

160 use_global: bool = global_option, 

161) -> None: 

162 """List installed models across all sources.""" 

163 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global) 

164 parsed_task = _parse_task_or_bad_param(task) 

165 data = list_models_data(source=_parse_source_or_bad_param(source), task=parsed_task) 

166 if cfg.json_mode: 

167 json_output(data.model_dump()) 

168 return 

169 if not data.models: 

170 console.print("No models installed.") 

171 return 

172 console.print(_render_list(data)) 

173 

174 

175@model_app.command("show") 

176def show_cmd( 

177 ref: str = typer.Argument(..., help="Model ref (e.g. 'Qwen/Qwen3-0.6B-GGUF')."), 

178 data_dir: Path | None = data_dir_option, 

179 use_global: bool = global_option, 

180) -> None: 

181 """Show catalog and installed metadata for a model.""" 

182 from lilbee.modelhub.model_manager import ModelNotFoundError 

183 

184 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global) 

185 try: 

186 data = show_model_data(ref) 

187 except ModelNotFoundError as exc: 

188 if cfg.json_mode: 

189 json_output({"error": str(exc)}) 

190 else: 

191 console.print(f"[{theme.ERROR}]{exc}[/{theme.ERROR}]") 

192 raise typer.Exit(1) from None 

193 if cfg.json_mode: 

194 json_output(data.model_dump()) 

195 return 

196 console.print(_render_show(data)) 

197 

198 

199def _run_pull( 

200 ref: str, 

201 src: ModelSource, 

202 on_update: Callable[[DownloadProgress], None], 

203 *, 

204 allow_unsupported: bool = False, 

205) -> PullResult: 

206 """Invoke ``pull_model_data`` and translate known errors to typer.Exit.""" 

207 from lilbee.catalog.compat import UnsupportedArchError 

208 

209 try: 

210 return pull_model_data(ref, src, on_update=on_update, allow_unsupported=allow_unsupported) 

211 except UnsupportedArchError as exc: 

212 msg = ( 

213 f"Architecture {exc.architecture!r} is not supported by this lilbee build.\n" 

214 "Pass --allow-unsupported to try anyway." 

215 ) 

216 if cfg.json_mode: 

217 json_output( 

218 { 

219 "error": "unsupported_arch", 

220 "arch": exc.architecture, 

221 "ref": exc.ref, 

222 } 

223 ) 

224 else: 

225 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {msg}") 

226 raise typer.Exit(1) from None 

227 except (RuntimeError, PermissionError) as exc: 

228 if cfg.json_mode: 

229 json_output({"error": str(exc)}) 

230 else: 

231 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {exc}") 

232 raise typer.Exit(1) from None 

233 

234 

235def _pull_json_stream(ref: str, src: ModelSource, *, allow_unsupported: bool) -> None: 

236 """Emit newline-delimited JSON progress events, then the final result.""" 

237 

238 def on_update(p: DownloadProgress) -> None: 

239 event = PullProgressEvent( 

240 model=ref, percent=p.percent, detail=p.detail, cache_hit=p.is_cache_hit 

241 ) 

242 json_output(event.model_dump()) 

243 

244 final = _run_pull(ref, src, on_update, allow_unsupported=allow_unsupported) 

245 json_output({**final.model_dump(), "event": PullEvent.DONE.value}) 

246 

247 

248def _pull_interactive_progress(ref: str, src: ModelSource, *, allow_unsupported: bool) -> None: 

249 """Drive Rich's Live progress bar during a native HuggingFace download.""" 

250 err_console = Console(stderr=True, force_terminal=True) 

251 with Progress( 

252 TextColumn("[progress.description]{task.description}"), 

253 BarColumn(), 

254 TextColumn("{task.percentage:>3.0f}%"), 

255 TextColumn("{task.fields[detail]}"), 

256 TimeRemainingColumn(), 

257 console=err_console, 

258 transient=False, 

259 ) as progress: 

260 task_id = progress.add_task(f"Downloading {ref}", total=100, detail="") 

261 

262 def on_update(p: DownloadProgress) -> None: 

263 progress.update(task_id, completed=p.percent, detail=p.detail) 

264 

265 final = _run_pull(ref, src, on_update, allow_unsupported=allow_unsupported) 

266 

267 if final.status == PullStatus.ALREADY_INSTALLED: 

268 console.print(f"{ref} is already installed.") 

269 else: 

270 console.print(f"Pulled [{theme.ACCENT}]{ref}[/{theme.ACCENT}].") 

271 

272 

273@model_app.command("pull") 

274def pull_cmd( 

275 ref: str = typer.Argument(..., help="Model ref to download (e.g. 'Qwen/Qwen3-0.6B-GGUF')."), 

276 source: str = typer.Option( 

277 "native", 

278 "--source", 

279 "-s", 

280 help="Pull from 'native' (HuggingFace GGUF) or 'remote' (SDK-managed).", 

281 ), 

282 allow_unsupported: bool = typer.Option( 

283 False, 

284 "--allow-unsupported", 

285 help="Pull even if the architecture isn't in the supported set (load may still fail).", 

286 ), 

287 data_dir: Path | None = data_dir_option, 

288 use_global: bool = global_option, 

289) -> None: 

290 """Download a model.""" 

291 from lilbee.catalog.types import ModelSource 

292 

293 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global) 

294 src = _parse_source_or_bad_param(source) or ModelSource.NATIVE 

295 if cfg.json_mode: 

296 _pull_json_stream(ref, src, allow_unsupported=allow_unsupported) 

297 else: 

298 _pull_interactive_progress(ref, src, allow_unsupported=allow_unsupported) 

299 

300 

301def _confirm_remove_or_exit(ref: str, yes: bool) -> None: 

302 if yes or cfg.json_mode: 

303 return 

304 if not typer.confirm(f"Remove {ref}?", default=False): 

305 console.print("Aborted.") 

306 raise typer.Exit(0) 

307 

308 

309@model_app.command("rm") 

310def rm_cmd( 

311 ref: str = typer.Argument(..., help="Model ref to remove."), 

312 source: str | None = _source_option, 

313 yes: bool = _yes_option, 

314 data_dir: Path | None = data_dir_option, 

315 use_global: bool = global_option, 

316) -> None: 

317 """Remove an installed model.""" 

318 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global) 

319 src = _parse_source_or_bad_param(source) 

320 _confirm_remove_or_exit(ref, yes) 

321 try: 

322 data = remove_model_data(ref, source=src) 

323 except ValueError as exc: 

324 if cfg.json_mode: 

325 json_output({"error": str(exc)}) 

326 else: 

327 console.print(f"[{theme.ERROR}]{exc}[/{theme.ERROR}]") 

328 raise typer.Exit(1) from None 

329 if cfg.json_mode: 

330 json_output(data.model_dump()) 

331 if not data.deleted: 

332 raise typer.Exit(1) 

333 return 

334 if not data.deleted: 

335 console.print(f"[{theme.WARNING}]Not found: {ref}[/{theme.WARNING}]") 

336 raise typer.Exit(1) 

337 suffix = f" ({data.freed_gb:.2f} GB freed)" if data.freed_gb else "" 

338 console.print(f"Removed [{theme.ACCENT}]{ref}[/{theme.ACCENT}]{suffix}.") 

339 

340 

341def _is_interactive_terminal() -> bool: 

342 """Return True when both stdin and stdout are connected to a TTY. 

343 

344 Extracted as a module-level helper so tests can patch it deterministically; 

345 CliRunner replaces ``sys.stdin`` during invoke which makes direct 

346 monkey-patching of ``sys.stdin.isatty`` unreliable. 

347 """ 

348 return sys.stdin.isatty() and sys.stdout.isatty() 

349 

350 

351@model_app.command("browse") 

352def browse_cmd( 

353 data_dir: Path | None = data_dir_option, 

354 use_global: bool = global_option, 

355) -> None: 

356 """Open the Textual TUI directly on the model catalog screen. 

357 

358 Exit codes follow the project convention: 2 for invalid flag 

359 combinations (``--json`` with an interactive-only command), 1 for 

360 runtime environment failures (no TTY). 

361 """ 

362 _apply_catalog_overrides(data_dir=data_dir, use_global=use_global) 

363 if cfg.json_mode: 

364 json_output({"error": "model browse is interactive, not available in --json mode"}) 

365 raise typer.Exit(2) 

366 if not _is_interactive_terminal(): 

367 console.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] model browse requires a terminal.") 

368 raise typer.Exit(1) 

369 

370 from lilbee.cli.tui import run_tui 

371 from lilbee.cli.tui.messages import CATALOG_VIEW 

372 

373 run_tui(initial_view=CATALOG_VIEW)