Coverage for src/lilbee/server/handlers/config.py: 100%

1"""Config read/update handlers for the HTTP server.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import copy 

7import functools 

8from typing import Any 

9 

10from pydantic_core import PydanticUndefined 

11 

12from lilbee.app.settings import ( 

13 SettingInfo, 

14 apply_settings_update, 

15 list_settings, 

16 profile_defaults, 

17 provider_reset_refused_message, 

18 requires_services_reset, 

19 reset_settings, 

20 setting_sources, 

21) 

22from lilbee.config_meta import ( 

23 MODEL_ROLE_FIELDS as _MODEL_ROLE_FIELDS, 

24) 

25from lilbee.config_meta import ( 

26 PUBLIC_CONFIG_FIELDS as _PUBLIC_CONFIG_FIELDS, 

27) 

28from lilbee.config_meta import ( 

29 WRITABLE_CONFIG_FIELDS, 

30) 

31from lilbee.core.config import Config, cfg 

32from lilbee.server.models import ( 

33 ConfigFieldSchema, 

34 ConfigResponse, 

35 ConfigSchemaResponse, 

36 ConfigSourcesResponse, 

37 ConfigUpdateResponse, 

38) 

39 

40 

41async def update_config(updates: dict[str, Any]) -> ConfigUpdateResponse: 

42 """Partial update of writable config fields. 

43 

44 Delegates validation, snapshot/rollback, persistence, and cache 

45 invalidation to ``app.settings.apply_settings_update`` so HTTP, MCP, 

46 CLI, and the TUI share one write boundary. Model role writes are 

47 refused at this surface because PUT /api/models/<role> already 

48 handles them with an install-availability check. 

49 

50 A provider switch rebuilds the shared Services singleton, unsafe while other 

51 clients have in-flight requests, so it is refused on the always-concurrent 

52 HTTP server; do it from the CLI instead. 

53 """ 

54 if requires_services_reset(updates): 

55 raise ValueError(provider_reset_refused_message("Switching")) 

56 result = await asyncio.to_thread(apply_settings_update, updates, allow_model_roles=False) 

57 return ConfigUpdateResponse.from_result(result) 

58 

59 

60async def reset_config(keys: list[str]) -> ConfigUpdateResponse: 

61 """Remove each key from config.toml so its value falls back to the next source. 

62 

63 Refuses the same keys PATCH /api/config refuses: a provider switch and the 

64 model role slots. 

65 """ 

66 if requires_services_reset(dict.fromkeys(keys)): 

67 raise ValueError(provider_reset_refused_message("Resetting")) 

68 result = await asyncio.to_thread(reset_settings, keys, allow_model_roles=False) 

69 return ConfigUpdateResponse.from_result(result) 

70 

71 

72async def get_config() -> ConfigResponse: 

73 """Return all user-facing configuration values.""" 

74 dumped = cfg.model_dump() 

75 result = {k: v for k, v in dumped.items() if k in _PUBLIC_CONFIG_FIELDS} 

76 return ConfigResponse(**result) 

77 

78 

79async def get_config_sources() -> ConfigSourcesResponse: 

80 """Return the source of every value GET /api/config answers.""" 

81 sources = await asyncio.to_thread(setting_sources) 

82 return ConfigSourcesResponse( 

83 sources={k: v for k, v in sources.items() if k in _PUBLIC_CONFIG_FIELDS} 

84 ) 

85 

86 

87@functools.cache 

88def _compute_config_defaults() -> dict[str, Any]: 

89 """Materialize Config defaults once per process.""" 

90 defaults: dict[str, Any] = {} 

91 for name, info in Config.model_fields.items(): 

92 # The writable conjunct is redundant today (every public field is 

93 # writable or a model role) but keeps a future public-but-not-writable 

94 # field out of this payload. 

95 is_writable_public = name in WRITABLE_CONFIG_FIELDS and name in _PUBLIC_CONFIG_FIELDS 

96 if not is_writable_public and name not in _MODEL_ROLE_FIELDS: 

97 continue 

98 value = info.get_default(call_default_factory=True) 

99 if value is PydanticUndefined: # pragma: no cover 

100 continue 

101 defaults[name] = value 

102 return defaults 

103 

104 

105async def get_config_defaults() -> ConfigResponse: 

106 """Return the default of every public config field. 

107 

108 Covers writable fields (resettable via PATCH /api/config) and the 

109 model-role fields (resettable via PUT /api/models/<role>). A value the 

110 active profile sets replaces the built-in, as in ``lilbee settings list``, 

111 because a reset falls back to it. 

112 

113 Deepcopies the cached dict so callers that mutate the response 

114 (list-valued fields like ``crawl_exclude_patterns``) cannot poison 

115 subsequent calls. 

116 """ 

117 defaults = copy.deepcopy(_compute_config_defaults()) 

118 profile = await asyncio.to_thread(profile_defaults) 

119 defaults.update({key: value for key, value in profile.items() if key in defaults}) 

120 return ConfigResponse(**defaults) 

121 

122 

123def _field_schema(info: SettingInfo) -> ConfigFieldSchema: 

124 """Render one setting's metadata for the wire. 

125 

126 ``writable`` is the PATCH /api/config contract, so it is false for the 

127 model role slots, which PUT /api/models/<role> owns. 

128 """ 

129 return ConfigFieldSchema( 

130 key=info.key, 

131 type=info.type, 

132 nullable=info.nullable, 

133 writable=info.key in WRITABLE_CONFIG_FIELDS, 

134 reindex_required=info.reindex_required, 

135 group=info.group, 

136 help=info.help_text, 

137 choices=list(info.choices) if info.choices else None, 

138 advanced=info.advanced, 

139 ) 

140 

141 

142async def get_config_schema() -> ConfigSchemaResponse: 

143 """Return per-field metadata for every public configuration field. 

144 

145 The field list comes from the same settings boundary that MCP 

146 ``settings_list`` reads, so a new setting appears here with no route 

147 change and no restated value set. 

148 """ 

149 infos = await asyncio.to_thread(list_settings) 

150 return ConfigSchemaResponse(fields=[_field_schema(info) for info in infos])