Coverage for src/lilbee/app/status.py: 100%
122 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Status snapshot of the local knowledge base."""
3from __future__ import annotations
5import os
6from pathlib import Path
8from pydantic import BaseModel
10from lilbee.app.services import get_services
11from lilbee.app.settings import scanned_pages_state
12from lilbee.core.config import cfg
13from lilbee.core.config.enums import KvCacheType, OcrMode
14from lilbee.core.system import LOCAL_ROOT_DIRNAME, default_data_dir
15from lilbee.data.store import HeldOut
16from lilbee.data.types import SkippedSource
18LILBEE_LABEL_MAX_LEN = 40
19"""Hard cap on the compact-label width. The leaf gets its own internal
20ellipsis when even the leaf alone would breach the cap."""
22STATUS_SKIPPED_LIMIT = 50
23"""Cap on held-out files in one status response; ``skipped_total`` carries the real count."""
25STATUS_RECENT_LIMIT = 10
26"""Cap on the newest sources in one status response; ``document_count`` carries the real count."""
28_ELLIPSIS = "…"
31def _project_root() -> Path:
32 """Walk past a trailing ``.lilbee`` marker to the project dir that owns it."""
33 root = cfg.data_root
34 if root.name == LOCAL_ROOT_DIRNAME:
35 return root.parent
36 return root
39def _truncate_leaf(leaf: str, max_len: int) -> str:
40 """Shrink an over-long leaf to fit a budget, with an internal ellipsis."""
41 if len(leaf) <= max_len:
42 return leaf
43 if max_len <= 1:
44 return _ELLIPSIS
45 keep = max_len - 1
46 head = keep // 2
47 tail = keep - head
48 return f"{leaf[:head]}{_ELLIPSIS}{leaf[-tail:] if tail else ''}"
51def _compact_path(full: str) -> str:
52 """Render *full* with ``~`` substituted for ``$HOME`` when it leads."""
53 home = str(Path.home())
54 if full == home:
55 return "~"
56 home_prefix = f"{home}{os.sep}"
57 return f"~{os.sep}{full[len(home_prefix) :]}" if full.startswith(home_prefix) else full
60def lilbee_label() -> str:
61 """Status-bar pill text for the active lilbee.
63 Precedence: ``lilbee_name`` override > ``"global"`` (when data_root
64 is the platform default) > project path. ``show_lilbee_path``
65 (toggled by F4) returns the full absolute path instead of the
66 compact / "global" form.
67 """
68 if cfg.lilbee_name:
69 return cfg.lilbee_name
70 is_global = cfg.data_root.expanduser().resolve() == default_data_dir().resolve()
71 if cfg.show_lilbee_path:
72 return str(default_data_dir() if is_global else _project_root().expanduser().resolve())
73 if is_global:
74 return "global"
75 full = str(_project_root().expanduser().resolve())
76 compact = _compact_path(full)
77 if len(compact) <= LILBEE_LABEL_MAX_LEN:
78 return compact
79 leaf = _project_root().name or compact
80 leaf_budget = LILBEE_LABEL_MAX_LEN - 1 - len(os.sep)
81 return f"{_ELLIPSIS}{os.sep}{_truncate_leaf(leaf, leaf_budget)}"
84class StatusConfig(BaseModel):
85 """Configuration section of a status response.
87 Exposes all four role-bound model fields (chat, embedding, vision,
88 reranker) so the TUI status screen and plugin callers can show
89 what's active per role.
90 """
92 documents_dir: str
93 data_dir: str
94 chat_model: str
95 embedding_model: str
96 vision_model: str = ""
97 reranker_model: str = ""
98 ocr: OcrMode = OcrMode.AUTO
99 num_ctx: int | None = None
100 num_ctx_max: int | None = None
101 chat_n_ctx_target: int | None = None
102 flash_attention: bool | None = None
103 kv_cache_type: KvCacheType | None = None
104 n_gpu_layers: int | None = None
105 cpu_moe: bool | None = None
106 n_cpu_moe: int | None = None
107 gpu_devices: str | None = None
110class IndexStatus(BaseModel):
111 """The embedder that built the persisted index."""
113 embedding_model: str
114 embedding_dim: int
117class SourceInfo(BaseModel):
118 """A single indexed source in a status response."""
120 filename: str
121 file_hash: str
122 chunk_count: int
123 ingested_at: str
126class EntityStatus(BaseModel):
127 """Entity-extraction section of a status response (present when enabled)."""
129 types: list[str]
130 rows: int
133class StatusResult(BaseModel):
134 """Full status response for the knowledge base."""
136 command: str = "status"
137 config: StatusConfig
138 sources: list[SourceInfo]
139 """The newest sources, at most ``STATUS_RECENT_LIMIT``; ``document_count`` is the real count."""
140 source_kinds: dict[str, int] = {}
141 """How many sources each kind holds, over the whole library."""
142 document_count: int
143 total_chunks: int
144 index: IndexStatus | None = None
145 """The embedder that built the index; None before the first sync."""
146 entities: EntityStatus | None = None
147 skipped: list[SkippedSource] = []
148 """Files a held-out record holds out of the index, capped at ``STATUS_SKIPPED_LIMIT``."""
149 skipped_total: int = 0
150 ocr_note: str = ""
151 """What happens to scanned pages: skipped, or read by which engine."""
154def _index_status() -> IndexStatus | None:
155 """The persisted index identity, or None when nothing has been indexed."""
156 meta = get_services().store.get_meta()
157 if meta is None:
158 return None
159 return IndexStatus(embedding_model=meta["embedding_model"], embedding_dim=meta["embedding_dim"])
162def gather_status() -> StatusResult:
163 """Collect status data as a typed model (shared by human + JSON output)."""
164 summary = get_services().store.source_summary(STATUS_RECENT_LIMIT)
165 skipped, skipped_total = held_out_sources()
166 return StatusResult(
167 index=_index_status(),
168 config=StatusConfig(
169 documents_dir=str(cfg.documents_dir),
170 data_dir=str(cfg.data_dir),
171 chat_model=cfg.chat_model,
172 embedding_model=cfg.embedding_model,
173 vision_model=cfg.vision_model,
174 reranker_model=cfg.reranker_model,
175 ocr=cfg.ocr,
176 num_ctx=cfg.num_ctx,
177 num_ctx_max=cfg.num_ctx_max,
178 chat_n_ctx_target=cfg.chat_n_ctx_target,
179 flash_attention=cfg.flash_attention,
180 kv_cache_type=cfg.kv_cache_type,
181 n_gpu_layers=cfg.n_gpu_layers,
182 cpu_moe=cfg.cpu_moe,
183 n_cpu_moe=cfg.n_cpu_moe,
184 gpu_devices=cfg.gpu_devices,
185 ),
186 sources=[
187 SourceInfo(
188 filename=head["filename"],
189 file_hash=head["file_hash"][:12],
190 chunk_count=head["chunk_count"],
191 ingested_at=head["ingested_at"][:19],
192 )
193 for head in summary.newest
194 ],
195 source_kinds=summary.kinds,
196 document_count=summary.document_count,
197 total_chunks=summary.total_chunks,
198 entities=entity_status(),
199 skipped=skipped,
200 skipped_total=skipped_total,
201 ocr_note=scanned_pages_state(),
202 )
205def held_out_sources() -> tuple[list[SkippedSource], int]:
206 """Failed files with reasons, capped at ``STATUS_SKIPPED_LIMIT``, and the real count."""
207 held = get_services().store.held_records()
208 failed = sorted(name for name, record in held.items() if record.kind is HeldOut.FAILED)
209 shown = [SkippedSource(filename=name, reason=held[name].reason) for name in failed]
210 return shown[:STATUS_SKIPPED_LIMIT], len(failed)
213def entity_status() -> EntityStatus | None:
214 """Entity types + extracted row count; None while the feature is off."""
215 if not cfg.entity_extraction:
216 return None
217 from lilbee.core.config import ENTITIES_TABLE
218 from lilbee.retrieval.entities import load_schema
220 store = get_services().store
221 schema = load_schema(store)
222 table = store.open_table(ENTITIES_TABLE)
223 rows = table.count_rows() if table is not None else 0
224 return EntityStatus(types=[t.name for t in schema.types] if schema else [], rows=rows)