Coverage for src/lilbee/core/settings.py: 100%

1"""Persistent settings stored in config.toml alongside the data directory.""" 

2 

3import logging 

4import os 

5import threading 

6import tomllib 

7from collections.abc import Callable, Generator 

8from contextlib import contextmanager 

9from pathlib import Path, PurePath 

10from typing import Any, TypeVar 

11 

12import tomli_w 

13 

14from lilbee.config_meta import MODEL_ROLE_FIELDS, WRITABLE_CONFIG_FIELDS 

15from lilbee.core.config import CONFIG_FILE_NAME, cfg 

16from lilbee.core.config.model import value_is_set 

17from lilbee.core.security import file_lock_or_warn, harden_private_file, write_private_text 

18 

19_settings_lock = threading.Lock() 

20 

21T = TypeVar("T") 

22 

23# A server, a CLI invocation, and an MCP process routinely run against the same 

24# data root, so the in-process mutex alone lets two of them interleave a 

25# read-modify-write and silently drop each other's keys. 

26_CONFIG_LOCK_TIMEOUT_S = 10.0 

27 

28 

29def _config_path(data_root: Path) -> Path: 

30 return data_root / CONFIG_FILE_NAME 

31 

32 

33@contextmanager 

34def _config_write_lock(data_root: Path) -> Generator[None, None, None]: 

35 """Serialize a config read-modify-write across threads and processes. 

36 

37 A lock timeout falls through to the write rather than failing the caller: 

38 losing a settings update to a stale lock file is worse than the interleave 

39 the lock exists to prevent, which is already rare. 

40 """ 

41 path = _config_path(data_root) 

42 path.parent.mkdir(parents=True, exist_ok=True) 

43 with _settings_lock, file_lock_or_warn(path, _CONFIG_LOCK_TIMEOUT_S): 

44 yield 

45 

46 

47def load(data_root: Path) -> dict[str, Any]: 

48 """Read all settings from config.toml. Returns {} if file is missing. 

49 

50 Values keep the types TOML gave them. Stringifying here used to turn a 

51 ``true`` into ``"True"`` in memory, which the next save then wrote back 

52 quoted, so the file drifted away from valid types for its own fields. 

53 """ 

54 path = _config_path(data_root) 

55 if not path.exists(): 

56 return {} 

57 harden_private_file(path) 

58 with path.open("rb") as f: 

59 return dict(tomllib.load(f)) 

60 

61 

62def save(data_root: Path, settings: dict[str, Any]) -> None: 

63 """Write *settings* to config.toml. 

64 

65 ``tomli_w`` is the write half of the stdlib ``tomllib`` used by ``load``. 

66 The emitter this replaced escaped strings by hand and stringified anything 

67 that was not a bool or a number, so a list value was persisted as its 

68 quoted repr and read back as text. A control character it escaped wrongly 

69 was worse still: the reader discards the whole file on a parse error, so 

70 one bad value silently wiped every other setting. 

71 

72 A ``None`` is dropped rather than written. TOML has no null, and the old 

73 emitter wrote the literal string "None", which then read back as a set 

74 value instead of an absent one. A path is written as its string: tomli_w 

75 refuses ``Path`` objects, and the config's path fields hold them after 

76 validation. 

77 """ 

78 path = _config_path(data_root) 

79 present = { 

80 k: str(v) if isinstance(v, PurePath) else v 

81 for k, v in sorted(settings.items()) 

82 if v is not None 

83 } 

84 # config.toml can hold provider API keys, so it gets the same owner-only 

85 # treatment as the session token rather than a post-hoc chmod. 

86 write_private_text(path, tomli_w.dumps(present)) 

87 

88 

89def get(data_root: Path, key: str) -> str | None: 

90 """Look up a single key from config.toml, as text for callers that want text.""" 

91 value = load(data_root).get(key) 

92 return None if value is None else str(value) 

93 

94 

95def set_value(data_root: Path, key: str, value: Any) -> None: 

96 """Read-modify-write a single key in config.toml.""" 

97 with _config_write_lock(data_root): 

98 current = load(data_root) 

99 current[key] = value 

100 save(data_root, current) 

101 

102 

103def delete_value(data_root: Path, key: str) -> None: 

104 """Remove a key from config.toml. No-op if key doesn't exist.""" 

105 with _config_write_lock(data_root): 

106 current = load(data_root) 

107 current.pop(key, None) 

108 save(data_root, current) 

109 

110 

111def update_values(data_root: Path, updates: dict[str, Any]) -> None: 

112 """Batch update multiple keys in config.toml (single write).""" 

113 with _config_write_lock(data_root): 

114 current = load(data_root) 

115 current.update(updates) 

116 save(data_root, current) 

117 

118 

119def delete_values(data_root: Path, keys: list[str]) -> None: 

120 """Batch delete multiple keys from config.toml (single write).""" 

121 with _config_write_lock(data_root): 

122 current = load(data_root) 

123 for key in keys: 

124 current.pop(key, None) 

125 save(data_root, current) 

126 

127 

128def mutate_value(data_root: Path, key: str, fn: Callable[[Any], tuple[Any, T]]) -> T: 

129 """Read-modify-write a single key under the config lock, atomically. 

130 

131 ``fn`` receives the key's persisted value (or None if absent) read *inside* 

132 the lock and returns ``(new_value, result)``; the new value is written and 

133 the result is returned. Unlike a read-then-:func:`set_value`, the whole 

134 compound update is serialized across threads and processes, so two callers 

135 updating a dict-valued key cannot lose each other's change. 

136 """ 

137 with _config_write_lock(data_root): 

138 current = load(data_root) 

139 new_value, result = fn(current.get(key)) 

140 current[key] = new_value 

141 save(data_root, current) 

142 return result 

143 

144 

145def overlay_persisted_settings(root: Path) -> None: 

146 """Overlay persisted scalars from ``<root>/config.toml`` onto cfg, skipping bad values. 

147 

148 An explicit ``LILBEE_<FIELD>`` env var wins over config.toml (the documented 

149 precedence): cfg already holds the env-loaded value, so a key whose env var is 

150 set is left untouched rather than overwritten by the persisted file. An empty 

151 persisted value is skipped, except on a clearable model role, where it clears it. 

152 

153 ``LILBEE_SKIP_TOML_CONFIG=1`` disables this overlay entirely, matching the 

154 pydantic-settings source in ``config/model.py`` so the escape hatch is honored 

155 on every config-read path (import-time load, CLI callback, MCP server). 

156 """ 

157 if os.environ.get("LILBEE_SKIP_TOML_CONFIG") == "1": 

158 return 

159 log = logging.getLogger(__name__) 

160 try: 

161 persisted = load(root) 

162 except (OSError, ValueError): 

163 log.warning("Failed to read %s/config.toml; using in-memory defaults", root) 

164 return 

165 if not persisted: 

166 return 

167 overlayable = set(WRITABLE_CONFIG_FIELDS) | set(MODEL_ROLE_FIELDS) 

168 env_prefix = cfg.model_config.get("env_prefix", "") 

169 for key, raw in persisted.items(): 

170 if key not in overlayable: 

171 continue 

172 if value_is_set(key, os.environ.get(f"{env_prefix}{key.upper()}")): 

173 continue 

174 if not value_is_set(key, raw): 

175 continue 

176 try: 

177 setattr(cfg, key, raw) 

178 except (ValueError, TypeError) as exc: 

179 log.warning( 

180 "Ignoring invalid persisted value for %s in %s: %s", 

181 key, 

182 root, 

183 exc, 

184 )