Coverage for src / lilbee / app / status.py: 100%

1"""Status snapshot of the local knowledge base.""" 

2 

3from __future__ import annotations 

4 

5from pydantic import BaseModel 

6 

7from lilbee.app.services import get_services 

8from lilbee.core.config import cfg 

9 

10 

11class StatusConfig(BaseModel): 

12 """Configuration section of a status response. 

13 

14 Exposes all four role-bound model fields (chat, embedding, vision, 

15 reranker) so the TUI status screen and plugin callers can show 

16 what's active per role. 

17 """ 

18 

19 documents_dir: str 

20 data_dir: str 

21 chat_model: str 

22 embedding_model: str 

23 vision_model: str = "" 

24 reranker_model: str = "" 

25 enable_ocr: bool | None = None 

26 

27 

28class SourceInfo(BaseModel): 

29 """A single indexed source in a status response.""" 

30 

31 filename: str 

32 file_hash: str 

33 chunk_count: int 

34 ingested_at: str 

35 

36 

37class StatusResult(BaseModel): 

38 """Full status response for the knowledge base.""" 

39 

40 command: str = "status" 

41 config: StatusConfig 

42 sources: list[SourceInfo] 

43 total_chunks: int 

44 

45 

46def gather_status() -> StatusResult: 

47 """Collect status data as a typed model (shared by human + JSON output).""" 

48 sources = get_services().store.get_sources() 

49 sorted_sources = sorted(sources, key=lambda x: x["filename"]) 

50 total_chunks = sum(s["chunk_count"] for s in sources) 

51 return StatusResult( 

52 config=StatusConfig( 

53 documents_dir=str(cfg.documents_dir), 

54 data_dir=str(cfg.data_dir), 

55 chat_model=cfg.chat_model, 

56 embedding_model=cfg.embedding_model, 

57 vision_model=cfg.vision_model, 

58 reranker_model=cfg.reranker_model, 

59 enable_ocr=cfg.enable_ocr, 

60 ), 

61 sources=[ 

62 SourceInfo( 

63 filename=s["filename"], 

64 file_hash=s["file_hash"][:12], 

65 chunk_count=s["chunk_count"], 

66 ingested_at=s["ingested_at"][:19], 

67 ) 

68 for s in sorted_sources 

69 ], 

70 total_chunks=total_chunks, 

71 )