Coverage for src/lilbee/app/settings.py: 100%
277 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-08 09:20 +0000
1"""Canonical write boundary for lilbee configuration."""
3from __future__ import annotations
5import errno
6import types
7from dataclasses import dataclass
8from pathlib import Path
9from typing import TYPE_CHECKING, Any, Union, get_args, get_origin
11from pydantic_core import PydanticUndefined
13from lilbee.app.settings_map import SETTINGS_MAP, SettingDef, SettingGroup
14from lilbee.config_meta import (
15 MODEL_ROLE_FIELDS,
16 REINDEX_FIELDS,
17 WRITABLE_CONFIG_FIELDS,
18)
19from lilbee.core import settings as persistent_settings
20from lilbee.core.config import CONFIG_FILE_NAME, Config, cfg
21from lilbee.core.config.keys import (
22 LOAD_AFFECTING_KEYS,
23 PROVIDER_API_KEYS,
24 PROVIDER_SWITCHING_KEYS,
25)
27if TYPE_CHECKING:
28 from lilbee.modelhub.registry import ModelRegistry
30_MIN_CHUNK_SIZE = 64
32# Path-typed writable fields whose pydantic "default" is the unresolved
33# sentinel ``Path()`` (a literal "."). The actual default is computed by
34# the model_validator at process start (data_root/documents, vault_base
35# stays as None). Resetting these via the boundary would corrupt the
36# install, so they are refused at the reset gate.
37_NO_RESET_FIELDS: frozenset[str] = frozenset({"documents_dir"})
40@dataclass(frozen=True)
41class SettingInfo:
42 """Externally-facing description of a single writable setting."""
44 key: str
45 value: Any
46 default: Any
47 type: str
48 nullable: bool
49 group: SettingGroup
50 help_text: str
51 choices: tuple[str, ...] | None
52 reindex_required: bool
55@dataclass(frozen=True)
56class SettingsUpdateResult:
57 """Outcome of an ``apply_settings_update`` call."""
59 updated: list[str]
60 reindex_required: bool
63_SCALAR_TYPE_NAMES: dict[type, str] = {
64 bool: "bool",
65 int: "int",
66 float: "float",
67 str: "str",
68 Path: "str",
69 type(None): "null",
70}
71_COLLECTION_ORIGINS = (list, frozenset, set, tuple)
72_UNION_ORIGINS = (Union, types.UnionType)
75def _annotation_name(annotation: Any) -> str:
76 """Render a pydantic field annotation as a short MCP-friendly type string."""
77 origin = get_origin(annotation)
78 if origin in _UNION_ORIGINS:
79 return "|".join(_annotation_name(a) for a in get_args(annotation))
80 scalar = _SCALAR_TYPE_NAMES.get(annotation)
81 if scalar is not None:
82 return scalar
83 if origin in _COLLECTION_ORIGINS:
84 return "list"
85 return getattr(annotation, "__name__", None) or str(annotation)
88def _setting_default(key: str) -> Any:
89 """Return the pydantic default for ``key``, or ``None`` if unset."""
90 info = Config.model_fields[key]
91 if info.default_factory is not None:
92 return info.default_factory() # type: ignore[call-arg]
93 if info.default is PydanticUndefined:
94 return None
95 return info.default
98def _is_write_only(key: str) -> bool:
99 """Return True for fields persisted but never read back (API keys, hf_token)."""
100 extra = Config.model_fields[key].json_schema_extra
101 if isinstance(extra, dict):
102 return bool(extra.get("write_only", False))
103 return False
106def _public_writable_keys() -> list[str]:
107 """Names of every writable config field minus write-only secrets."""
108 keys = set(WRITABLE_CONFIG_FIELDS) | set(MODEL_ROLE_FIELDS)
109 return sorted(k for k in keys if not _is_write_only(k))
112def _setting_info(key: str, definition: SettingDef | None) -> SettingInfo:
113 field_info = Config.model_fields[key]
114 nullable = _is_nullable(key)
115 group = definition.group if definition else SettingGroup.MODELS
116 help_text = definition.help_text if definition else ""
117 choices = definition.choices if definition else None
118 return SettingInfo(
119 key=key,
120 value=getattr(cfg, key),
121 default=_setting_default(key),
122 type=_annotation_name(field_info.annotation),
123 nullable=nullable,
124 group=group,
125 help_text=help_text,
126 choices=choices,
127 reindex_required=key in REINDEX_FIELDS,
128 )
131def _parse_group(group: SettingGroup | str) -> SettingGroup:
132 """Resolve a group value or label to a ``SettingGroup``. Case-insensitive on the value."""
133 if isinstance(group, SettingGroup):
134 return group
135 normalized = group.strip().lower()
136 for candidate in SettingGroup:
137 if candidate.value.lower() == normalized:
138 return candidate
139 raise ValueError(
140 f"Unknown setting group: {group!r}. Valid groups: "
141 f"{', '.join(g.value for g in SettingGroup)}"
142 )
145def list_settings(group: SettingGroup | str | None = None) -> list[SettingInfo]:
146 """List every writable non-secret setting, optionally filtered by group (case-insensitive)."""
147 infos = [_setting_info(key, SETTINGS_MAP.get(key)) for key in _public_writable_keys()]
148 if group is not None:
149 wanted = _parse_group(group)
150 infos = [info for info in infos if info.group == wanted]
151 return sorted(infos, key=lambda info: (info.group.value, info.key))
154def get_setting(key: str) -> SettingInfo:
155 """Return the ``SettingInfo`` for one writable non-secret key."""
156 if not _is_settable(key):
157 raise KeyError(f"Unknown or read-only setting: {key}")
158 if _is_write_only(key):
159 raise KeyError(f"Setting '{key}' is write-only and cannot be read back")
160 return _setting_info(key, SETTINGS_MAP.get(key))
163def _is_settable(key: str) -> bool:
164 return key in WRITABLE_CONFIG_FIELDS or key in MODEL_ROLE_FIELDS
167def _is_nullable(key: str) -> bool:
168 """Return True if ``key`` accepts ``None`` to clear the persisted entry."""
169 if key in WRITABLE_CONFIG_FIELDS:
170 return WRITABLE_CONFIG_FIELDS[key]
171 return False
174def _as_int_setting(value: Any) -> int | None:
175 """Coerce a settings value to int the way pydantic will, or None if not numeric.
177 MCP settings_set forwards raw JSON, so a numeric setting can arrive as a
178 string (``{"chunk_overlap": "1000"}``). The cross-field guards must compare
179 the coerced int, not skip on the string and let pydantic accept an
180 unvalidated value downstream. ``bool`` is excluded (it is not a meaningful
181 chunk size) and non-numeric strings fall through to pydantic's type error.
182 """
183 if isinstance(value, bool):
184 return None
185 if isinstance(value, int):
186 return value
187 if isinstance(value, str):
188 try:
189 return int(value.strip())
190 except ValueError:
191 return None
192 return None
195def _validate(updates: dict[str, Any]) -> None:
196 """Reject unknown keys, null on non-nullable, and out-of-range chunk sizes."""
197 for key, value in updates.items():
198 if not _is_settable(key):
199 raise ValueError(f"Unknown or read-only setting: {key}")
200 if value is None and not _is_nullable(key):
201 raise ValueError(f"Setting '{key}' does not accept null")
202 new_ttl = _as_int_setting(updates.get("engine_idle_ttl_minutes"))
203 if new_ttl is not None and new_ttl < 0:
204 raise ValueError("engine_idle_ttl_minutes must be >= 0 (0 keeps weights loaded)")
205 new_chunk_size = _as_int_setting(updates.get("chunk_size"))
206 if new_chunk_size is not None and new_chunk_size < _MIN_CHUNK_SIZE:
207 raise ValueError(f"chunk_size must be >= {_MIN_CHUNK_SIZE}")
208 effective_chunk_size = new_chunk_size if new_chunk_size is not None else cfg.chunk_size
209 new_overlap = _as_int_setting(updates.get("chunk_overlap"))
210 # Compare the effective overlap against the effective chunk_size so that
211 # lowering chunk_size alone (below the already-persisted overlap) is caught,
212 # not just an explicit new overlap.
213 effective_overlap = new_overlap if new_overlap is not None else cfg.chunk_overlap
214 if effective_overlap >= effective_chunk_size:
215 raise ValueError(
216 f"chunk_overlap ({effective_overlap}) must be < chunk_size ({effective_chunk_size})"
217 )
220def _coerce_value(key: str, value: Any) -> Any:
221 """Canonicalize value before cfg assignment; model-role slots run task validation."""
222 if key in MODEL_ROLE_FIELDS and isinstance(value, str):
223 # heavy: role_validator pulls catalog + modelhub transitively (~300 ms)
224 from lilbee.modelhub.role_validator import validate_model_task_assignment
226 return validate_model_task_assignment(key, value)
227 return value
230def _apply_with_rollback(
231 updates: dict[str, Any],
232) -> tuple[dict[str, Any], list[str], dict[str, Any]]:
233 """Set each key on cfg with snapshot/rollback. Returns (persist, delete, snapshot)."""
234 snapshot = {k: getattr(cfg, k) for k in updates}
235 to_persist: dict[str, Any] = {}
236 to_delete: list[str] = []
237 try:
238 for key, raw in updates.items():
239 if raw is None:
240 setattr(cfg, key, None)
241 to_delete.append(key)
242 continue
243 setattr(cfg, key, _coerce_value(key, raw))
244 normalized = getattr(cfg, key)
245 if isinstance(normalized, list):
246 to_persist[key] = "\n".join(str(x) for x in normalized)
247 else:
248 # Hand the scalar over with its type intact so config.toml holds
249 # `true` and `2560`, not `"True"` and `"2560"`.
250 to_persist[key] = normalized
251 except Exception:
252 _restore_snapshot(snapshot)
253 raise
254 return to_persist, to_delete, snapshot
257def _restore_snapshot(snapshot: dict[str, Any]) -> None:
258 for key, value in snapshot.items():
259 setattr(cfg, key, value)
262def _reload_changed_roles(changed_keys: set[str]) -> None:
263 """Off-thread reload for each changed model-role server; full off-thread drop otherwise.
265 A model-role change (chat_model/embedding_model/reranker_model/vision_model)
266 respawns only that role's server via the per-role reload, so unrelated roles
267 keep serving uninterrupted. A genuinely role-agnostic load key (num_ctx,
268 kv_cache_type) has no single owning role, so it falls back to dropping the
269 whole fleet. Both paths run off the caller's thread, so the settings write
270 never blocks on a slow stop-and-respawn.
271 """
272 from lilbee.app.services import peek_services
273 from lilbee.providers.roles import MODEL_FIELD_TO_ROLE
275 services = peek_services()
276 if services is None:
277 return
278 changed_role_fields = changed_keys & MODEL_ROLE_FIELDS
279 for field in changed_role_fields:
280 services.reload_role(MODEL_FIELD_TO_ROLE[field])
281 if "vision_model" in changed_role_fields:
282 # Register/unregister lilbee's xberg OCR backend on any vision-model
283 # change (REST/MCP/TUI/CLI all funnel here), not just the REST route.
284 from lilbee.data.extract.backends import BackendKind, sync_xberg_backend
286 sync_xberg_backend(BackendKind.OCR, services.provider)
287 role_agnostic = (changed_keys & LOAD_AFFECTING_KEYS) - MODEL_ROLE_FIELDS
288 if role_agnostic:
289 services.provider.drop_loaded_models_async()
292def requires_services_reset(updates: dict[str, Any]) -> bool:
293 """True if applying *updates* would tear down and rebuild the Services singleton.
295 A provider switch reconstructs the provider via ``create_provider``, which
296 only runs at services init, so it forces a full ``reset_services()``. Callers
297 on the shared HTTP daemon use this to refuse the swap rather than tear the
298 singleton down under concurrent in-flight handlers.
299 """
300 return bool(set(updates) & PROVIDER_SWITCHING_KEYS)
303def provider_reset_refused_message(action: str) -> str:
304 """Shared user-facing refusal for a provider *action* on the HTTP server.
306 *action* is the verb shown to the user, e.g. ``"Switching"`` or
307 ``"Resetting"``. Kept in one place so the daemon entry points (MCP
308 settings_set / settings_reset, REST config) cannot drift apart.
309 """
310 return (
311 f"{action} the model provider is unavailable on the HTTP server: it rebuilds "
312 "the shared engine for every connected client. Change it from the CLI."
313 )
316def config_write_failure_message(exc: OSError) -> str:
317 """User-facing text for a failed config write; names the fix when the file is locked."""
318 detail = f"Could not write {CONFIG_FILE_NAME}: {exc}."
319 if exc.errno in (errno.EACCES, errno.EPERM):
320 detail += f" Close the program that holds {CONFIG_FILE_NAME} open and try again."
321 return detail
324def _invalidate_caches(changed_keys: set[str]) -> None:
325 """Drop every read-side cache whose freshness depends on a changed setting."""
326 if not changed_keys:
327 return
328 if changed_keys & MODEL_ROLE_FIELDS:
329 # heavy: model_info reads GGUF headers with the gguf parser (~130 ms)
330 from lilbee.modelhub.model_info import invalidate_cache as invalidate_arch_cache
332 invalidate_arch_cache()
333 if changed_keys & LOAD_AFFECTING_KEYS:
334 # heavy: app.services pulls the provider stack + lancedb (~70 ms)
335 _reload_changed_roles(changed_keys)
336 if "token_sizing" in changed_keys:
337 # Register/unregister lilbee's xberg tokenizer backend when token_sizing is
338 # toggled (via any settings path), so chunk sizing picks up the change
339 # without waiting for a services rebuild.
340 from lilbee.app.services import peek_services
341 from lilbee.data.extract.backends import BackendKind, sync_xberg_backend
343 services = peek_services()
344 if services is not None:
345 sync_xberg_backend(BackendKind.TOKENIZER, services.provider)
346 if changed_keys & PROVIDER_API_KEYS:
347 # heavy: sdk_llm_provider pulls litellm fanout (~145 ms)
348 from lilbee.providers.sdk_llm_provider import inject_provider_keys
350 inject_provider_keys()
351 if changed_keys & PROVIDER_SWITCHING_KEYS:
352 # Swap requires reconstructing the provider singleton via
353 # providers.factory.create_provider, only called at services init.
354 from lilbee.app.services import reset_services
356 reset_services()
357 if "mcp_tool_threads" in changed_keys:
358 # Resize the running server's thread pool now instead of only at startup.
359 from lilbee.server.app import reapply_thread_pool_ceiling
361 reapply_thread_pool_ceiling()
364def apply_settings_update(
365 updates: dict[str, Any],
366 *,
367 allow_model_roles: bool = True,
368) -> SettingsUpdateResult:
369 """Validate, apply, persist, and invalidate caches for a batch of updates.
371 Atomic on validation: a rejection rolls every field back and writes
372 nothing. Atomic on disk failure: an ``OSError`` from the TOML write, or a
373 parse error reloading a corrupt config.toml, restores the in-memory
374 snapshot before re-raising. Cache invalidation runs only after a
375 successful persist.
377 Pass ``allow_model_roles=False`` to reject ``chat_model`` /
378 ``embedding_model`` / ``vision_model`` / ``reranker_model`` at the
379 boundary; the HTTP PATCH /api/config surface uses this to route role
380 writes through PUT /api/models/<role>.
381 """
382 if not allow_model_roles:
383 rejected = MODEL_ROLE_FIELDS & set(updates)
384 if rejected:
385 offender = sorted(rejected)[0]
386 raise ValueError(
387 f"'{offender}' must be set through the dedicated model route, "
388 "not the general settings update."
389 )
390 _validate(updates)
391 embed_in_batch = "embedding_model" in updates
392 # Derived (not user-writable) fields applied alongside the validated batch.
393 effective_updates = dict(updates)
394 if embed_in_batch:
395 # Pin the OLD ref into store meta before mutation, otherwise the
396 # next read lazy-initializes meta from the NEW cfg and silently
397 # hides the dimension drift. Runs even when the value is unchanged
398 # so a legacy meta row is always canonicalized on the first swap
399 # attempt.
400 _pin_legacy_store_meta()
401 # Track the new embedder's output width so a fresh index is built at
402 # the right dimension (embedding_dim is derived, not in SETTINGS_MAP).
403 dim = _embedder_dim_from_gguf(updates["embedding_model"])
404 if dim is not None:
405 effective_updates["embedding_dim"] = dim
406 to_persist, to_delete, snapshot = _apply_with_rollback(effective_updates)
407 # embedding_dim is derived and applied to cfg in-memory, but the overlay
408 # loader ignores it on reload (it is re-derived), so don't write it to disk.
409 to_persist.pop("embedding_dim", None)
410 try:
411 if to_persist:
412 persistent_settings.update_values(cfg.data_root, to_persist)
413 if to_delete:
414 persistent_settings.delete_values(cfg.data_root, to_delete)
415 except (OSError, ValueError):
416 # OSError from the write, or a TOMLDecodeError (ValueError) when
417 # update/delete reloads a corrupt on-disk config.toml: either way the
418 # in-memory snapshot must be restored so cfg matches what was persisted.
419 _restore_snapshot(snapshot)
420 raise
421 _invalidate_caches(set(effective_updates))
422 reindex_required = bool(REINDEX_FIELDS & set(updates))
423 if embed_in_batch:
424 reindex_required = reindex_required or _embed_reindex_required(updates["embedding_model"])
425 return SettingsUpdateResult(
426 updated=sorted(updates),
427 reindex_required=reindex_required,
428 )
431def apply_ephemeral_model_swap(field: str, ref: str) -> None:
432 """Apply a chat/embedding model swap to cfg for this process only.
434 Performs the same embedding side effects as the persisted path (legacy
435 store meta pinned under the OLD ref first, then embedding_dim re-derived
436 for the new one) so the mismatch gate and table width stay correct, but
437 never writes config.toml.
438 """
439 if field == "embedding_model":
440 _pin_legacy_store_meta()
441 dim = _embedder_dim_from_gguf(ref)
442 setattr(cfg, field, ref)
443 if dim is not None:
444 cfg.embedding_dim = dim
445 return
446 setattr(cfg, field, ref)
449def _pin_legacy_store_meta() -> None:
450 """Pin the current embedding ref into store meta before swapping it."""
451 # heavy: ~100ms (lance + store init); only paid when embedding_model is in the batch.
452 from lilbee.app.services import get_services
454 get_services().store.initialize_meta_if_legacy()
457def _embedder_dim_from_gguf(ref: str, registry: ModelRegistry | None = None) -> int | None:
458 """The embedder's output width from its GGUF header (``<arch>.embedding_length``).
460 None when the model can't be resolved or the header lacks the field. Cheap: a
461 cached header read, no load. *registry* is forwarded to resolve the GGUF without
462 ``get_services()`` (callers running inside its construction).
463 """
464 from lilbee.providers.base import ProviderError
465 from lilbee.providers.engine_params import resolve_model_path
466 from lilbee.providers.gguf_meta import read_gguf_metadata
468 try:
469 # resolve_model_path raises ProviderError for a non-native (ollama/SDK) ref,
470 # which has no local GGUF -- those embedders carry no width to derive here.
471 meta = read_gguf_metadata(resolve_model_path(ref, registry))
472 except (ProviderError, ValueError, OSError, RuntimeError, TypeError):
473 return None
474 raw = meta.get("embedding_length") if meta else None
475 if not raw:
476 return None
477 try:
478 dim = int(raw)
479 except (TypeError, ValueError):
480 return None
481 return dim if dim > 0 else None
484def reconcile_embedding_dim(registry: ModelRegistry | None = None) -> None:
485 """Pin ``cfg.embedding_dim`` to the native embedder's GGUF width before the store
486 is built; no-op for non-native embedders or an already-matching dim."""
487 dim = _embedder_dim_from_gguf(cfg.embedding_model, registry)
488 if dim is not None and dim != cfg.embedding_dim:
489 cfg.embedding_dim = dim
492def _embed_reindex_required(new_ref: str) -> bool:
493 """Compare *new_ref* to the persisted store meta; True if rebuild needed."""
494 from lilbee.app.services import get_services
495 from lilbee.data.store.lance_helpers import refs_compatible
497 store = get_services().store
498 store.canonicalize_meta_if_legacy()
499 meta = store.get_meta()
500 if meta is None:
501 return False
502 return not refs_compatible(
503 meta["embedding_model"], new_ref, meta["embedding_dim"], meta["embedding_dim"]
504 )
507def reset_settings(keys: list[str], *, skip_unresettable: bool = False) -> SettingsUpdateResult:
508 """Reset each key to its pydantic default and apply through the write boundary.
510 Fields whose default is a known sentinel (currently ``documents_dir``,
511 which resolves to ``data_root/documents`` at process start) are
512 refused so a reset doesn't write the literal sentinel back. Pass
513 ``skip_unresettable=True`` for bulk-reset gestures that should drop
514 those fields rather than failing the whole batch.
515 """
516 for key in keys:
517 if not _is_settable(key):
518 raise ValueError(f"Unknown or read-only setting: {key}")
519 if key in _NO_RESET_FIELDS and not skip_unresettable:
520 raise ValueError(
521 f"'{key}' has no resettable default; pass an explicit value via settings_set."
522 )
523 updates: dict[str, Any] = {}
524 for key in keys:
525 if key in _NO_RESET_FIELDS:
526 continue
527 default = _setting_default(key)
528 if default is None and _is_nullable(key):
529 updates[key] = None
530 else:
531 updates[key] = default
532 return apply_settings_update(updates)