Coverage for src/lilbee/app/status.py: 100%
94 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-08 09:20 +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.core.config import cfg
12from lilbee.core.system import LOCAL_ROOT_DIRNAME, default_data_dir
13from lilbee.data.ingest.skip_marker import describe_skips, load_skip_markers
14from lilbee.data.types import SkippedSource
16LILBEE_LABEL_MAX_LEN = 40
17"""Hard cap on the compact-label width. The leaf gets its own internal
18ellipsis when even the leaf alone would breach the cap."""
20STATUS_SKIPPED_LIMIT = 50
21"""Cap on held-out files in one status response; ``skipped_total`` carries the real count."""
23_ELLIPSIS = "…"
26def _project_root() -> Path:
27 """Walk past a trailing ``.lilbee`` marker to the project dir that owns it."""
28 root = cfg.data_root
29 if root.name == LOCAL_ROOT_DIRNAME:
30 return root.parent
31 return root
34def _truncate_leaf(leaf: str, max_len: int) -> str:
35 """Shrink an over-long leaf to fit a budget, with an internal ellipsis."""
36 if len(leaf) <= max_len:
37 return leaf
38 if max_len <= 1:
39 return _ELLIPSIS
40 keep = max_len - 1
41 head = keep // 2
42 tail = keep - head
43 return f"{leaf[:head]}{_ELLIPSIS}{leaf[-tail:] if tail else ''}"
46def _compact_path(full: str) -> str:
47 """Render *full* with ``~`` substituted for ``$HOME`` when it leads."""
48 home = str(Path.home())
49 if full == home:
50 return "~"
51 home_prefix = f"{home}{os.sep}"
52 return f"~{os.sep}{full[len(home_prefix) :]}" if full.startswith(home_prefix) else full
55def lilbee_label() -> str:
56 """Status-bar pill text for the active lilbee.
58 Precedence: ``lilbee_name`` override > ``"global"`` (when data_root
59 is the platform default) > project path. ``show_lilbee_path``
60 (toggled by F4) returns the full absolute path instead of the
61 compact / "global" form.
62 """
63 if cfg.lilbee_name:
64 return cfg.lilbee_name
65 is_global = cfg.data_root.expanduser().resolve() == default_data_dir().resolve()
66 if cfg.show_lilbee_path:
67 return str(default_data_dir() if is_global else _project_root().expanduser().resolve())
68 if is_global:
69 return "global"
70 full = str(_project_root().expanduser().resolve())
71 compact = _compact_path(full)
72 if len(compact) <= LILBEE_LABEL_MAX_LEN:
73 return compact
74 leaf = _project_root().name or compact
75 leaf_budget = LILBEE_LABEL_MAX_LEN - 1 - len(os.sep)
76 return f"{_ELLIPSIS}{os.sep}{_truncate_leaf(leaf, leaf_budget)}"
79class StatusConfig(BaseModel):
80 """Configuration section of a status response.
82 Exposes all four role-bound model fields (chat, embedding, vision,
83 reranker) so the TUI status screen and plugin callers can show
84 what's active per role.
85 """
87 documents_dir: str
88 data_dir: str
89 chat_model: str
90 embedding_model: str
91 vision_model: str = ""
92 reranker_model: str = ""
93 enable_ocr: bool | None = None
96class SourceInfo(BaseModel):
97 """A single indexed source in a status response."""
99 filename: str
100 file_hash: str
101 chunk_count: int
102 ingested_at: str
105class EntityStatus(BaseModel):
106 """Entity-extraction section of a status response (present when enabled)."""
108 types: list[str]
109 rows: int
112class StatusResult(BaseModel):
113 """Full status response for the knowledge base."""
115 command: str = "status"
116 config: StatusConfig
117 sources: list[SourceInfo]
118 document_count: int
119 total_chunks: int
120 entities: EntityStatus | None = None
121 skipped: list[SkippedSource] = []
122 """Files a skip marker holds out of the index, capped at ``STATUS_SKIPPED_LIMIT``."""
123 skipped_total: int = 0
126def gather_status() -> StatusResult:
127 """Collect status data as a typed model (shared by human + JSON output)."""
128 sources = get_services().store.get_sources()
129 sorted_sources = sorted(sources, key=lambda x: x["filename"])
130 total_chunks = sum(s["chunk_count"] for s in sources)
131 skipped, skipped_total = held_out_sources()
132 return StatusResult(
133 config=StatusConfig(
134 documents_dir=str(cfg.documents_dir),
135 data_dir=str(cfg.data_dir),
136 chat_model=cfg.chat_model,
137 embedding_model=cfg.embedding_model,
138 vision_model=cfg.vision_model,
139 reranker_model=cfg.reranker_model,
140 enable_ocr=cfg.enable_ocr,
141 ),
142 sources=[
143 SourceInfo(
144 filename=s["filename"],
145 file_hash=s["file_hash"][:12],
146 chunk_count=s["chunk_count"],
147 ingested_at=s["ingested_at"][:19],
148 )
149 for s in sorted_sources
150 ],
151 document_count=len(sources),
152 total_chunks=total_chunks,
153 entities=entity_status(),
154 skipped=skipped,
155 skipped_total=skipped_total,
156 )
159def held_out_sources() -> tuple[list[SkippedSource], int]:
160 """Held-out files with reasons, capped at ``STATUS_SKIPPED_LIMIT``, and the real count."""
161 held_out = sorted(load_skip_markers(cfg.data_root))
162 return describe_skips(cfg.data_root, held_out[:STATUS_SKIPPED_LIMIT]), len(held_out)
165def entity_status() -> EntityStatus | None:
166 """Entity types + extracted row count; None while the feature is off."""
167 if not cfg.entity_extraction:
168 return None
169 from lilbee.core.config import ENTITIES_TABLE
170 from lilbee.retrieval.entities import load_schema
172 store = get_services().store
173 schema = load_schema(store)
174 table = store.open_table(ENTITIES_TABLE)
175 rows = table.count_rows() if table is not None else 0
176 return EntityStatus(types=[t.name for t in schema.types] if schema else [], rows=rows)