Coverage for src / lilbee / core / config / validators.py: 100%

1"""Catalog/role validation helpers and the :func:`ConfigField` wrapper.""" 

2 

3import os 

4import sys 

5from typing import TYPE_CHECKING, Any 

6 

7from pydantic import Field 

8 

9from lilbee.providers.model_ref import PROVIDER_PREFIXES 

10 

11if TYPE_CHECKING: 

12 from lilbee.catalog.types import ModelTask 

13 

14 

15def ConfigField( # noqa: N802 pydantic Field wrapper; matches Field's PascalCase 

16 *args: Any, 

17 writable: bool = False, 

18 reindex: bool = False, 

19 write_only: bool = False, 

20 public: bool = True, 

21 **kwargs: Any, 

22) -> Any: 

23 """Wrap pydantic ``Field`` and attach metadata via ``json_schema_extra``.""" 

24 extra: dict[str, bool] = {} 

25 if writable: 

26 extra["writable"] = True 

27 if reindex: 

28 extra["reindex"] = True 

29 if write_only: 

30 extra["write_only"] = True 

31 if not public: 

32 extra["public"] = False 

33 if extra: 

34 kwargs["json_schema_extra"] = extra 

35 return Field(*args, **kwargs) 

36 

37 

38# Test-only bypass. Both the env var and pytest must be present so a 

39# leaked env var cannot disable validation in production. 

40_SKIP_MODEL_TASK_VALIDATION_ENV = "LILBEE_SKIP_MODEL_TASK_VALIDATION" 

41 

42 

43def _model_task_validation_bypassed() -> bool: 

44 if not os.environ.get(_SKIP_MODEL_TASK_VALIDATION_ENV): 

45 return False 

46 return sys.modules.get("pytest") is not None 

47 

48 

49_MODEL_FIELD_TO_TASK: dict[str, str] = { 

50 "chat_model": "chat", 

51 "embedding_model": "embedding", 

52 "vision_model": "vision", 

53 "reranker_model": "rerank", 

54} 

55 

56 

57class TaskMismatchError(ValueError): 

58 """A role slot was assigned a model whose catalog task does not match. 

59 

60 Carries the structured fields so each surface (HTTP, CLI, TUI, MCP) 

61 can format its own user-facing message. The default ``str()`` form is 

62 surface-neutral so it is safe to surface unmodified. 

63 """ 

64 

65 def __init__(self, ref: str, entry_task: "ModelTask", expected_task: "ModelTask") -> None: 

66 self.ref = ref 

67 self.entry_task = entry_task 

68 self.expected_task = expected_task 

69 super().__init__(f"Model '{ref}' is a {entry_task} model, not {expected_task}.") 

70 

71 

72# A native GGUF ref of the form ``<owner>/<repo>/<file>.gguf`` has at least 

73# two ``/`` separators; one-slash refs are bare repo IDs. 

74_NATIVE_GGUF_REF_MIN_SLASHES = 2 

75 

76 

77def _find_model_catalog_entry(ref: str) -> Any: 

78 # circular import: catalog imports cfg. 

79 from lilbee.catalog import find_catalog_entry 

80 

81 return find_catalog_entry(ref) 

82 

83 

84def _enforce_role_match(ref: str, entry: Any, field_name: str) -> None: 

85 from lilbee.catalog.types import ModelTask 

86 

87 want = ModelTask(_MODEL_FIELD_TO_TASK[field_name]) 

88 if entry.task == want: 

89 return 

90 raise TaskMismatchError(ref, ModelTask(entry.task), want) 

91 

92 

93def _skips_catalog_check(ref: str, *, allow_bypass: bool) -> bool: 

94 """Whether *ref* skips the catalog check.""" 

95 if not ref or not ref.strip(): 

96 return True 

97 if allow_bypass and _model_task_validation_bypassed(): 

98 return True 

99 return ref.split("/", 1)[0] in PROVIDER_PREFIXES 

100 

101 

102def validate_model_task_assignment(field_name: str, ref: str, *, allow_bypass: bool = True) -> str: 

103 """Check *ref* is a catalog entry whose task matches *field_name*; return the canonical ref.""" 

104 if _skips_catalog_check(ref, allow_bypass=allow_bypass): 

105 return ref 

106 entry = _find_model_catalog_entry(ref) 

107 if entry is None: 

108 raise ValueError( 

109 f"Model '{ref}' is not in the featured catalog. " 

110 "Pick a featured model for this role, or install one with " 

111 "'lilbee model pull <ref>' (or POST /api/models/pull) using a known catalog ref." 

112 ) 

113 _enforce_role_match(ref, entry, field_name) 

114 # Keep a full ``<repo>/<file>.gguf`` so resolve_model_path lands on 

115 # the exact installed quant; fall back to the catalog ref otherwise. 

116 if ref.endswith(".gguf") and ref.count("/") >= _NATIVE_GGUF_REF_MIN_SLASHES: 

117 return ref 

118 canonical: str = entry.ref 

119 return canonical