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.security import file_lock_or_warn, harden_private_file, write_private_text 

17 

18_settings_lock = threading.Lock() 

19 

20T = TypeVar("T") 

21 

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

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

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

25_CONFIG_LOCK_TIMEOUT_S = 10.0 

26 

27 

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

29 return data_root / CONFIG_FILE_NAME 

30 

31 

32@contextmanager 

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

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

35 

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

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

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

39 """ 

40 path = _config_path(data_root) 

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

42 with _settings_lock, file_lock_or_warn(path, _CONFIG_LOCK_TIMEOUT_S): 

43 yield 

44 

45 

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

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

48 

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

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

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

52 """ 

53 path = _config_path(data_root) 

54 if not path.exists(): 

55 return {} 

56 harden_private_file(path) 

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

58 return dict(tomllib.load(f)) 

59 

60 

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

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

63 

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

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

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

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

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

69 one bad value silently wiped every other setting. 

70 

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

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

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

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

75 validation. 

76 """ 

77 path = _config_path(data_root) 

78 present = { 

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

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

81 if v is not None 

82 } 

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

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

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

86 

87 

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

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

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

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

92 

93 

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

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

96 with _config_write_lock(data_root): 

97 current = load(data_root) 

98 current[key] = value 

99 save(data_root, current) 

100 

101 

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

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

104 with _config_write_lock(data_root): 

105 current = load(data_root) 

106 current.pop(key, None) 

107 save(data_root, current) 

108 

109 

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

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

112 with _config_write_lock(data_root): 

113 current = load(data_root) 

114 current.update(updates) 

115 save(data_root, current) 

116 

117 

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

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

120 with _config_write_lock(data_root): 

121 current = load(data_root) 

122 for key in keys: 

123 current.pop(key, None) 

124 save(data_root, current) 

125 

126 

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

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

129 

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

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

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

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

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

135 """ 

136 with _config_write_lock(data_root): 

137 current = load(data_root) 

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

139 current[key] = new_value 

140 save(data_root, current) 

141 return result 

142 

143 

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

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

146 

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

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

149 set is left untouched rather than overwritten by the persisted file. 

150 

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

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

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

154 """ 

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

156 return 

157 log = logging.getLogger(__name__) 

158 try: 

159 persisted = load(root) 

160 except (OSError, ValueError): 

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

162 return 

163 if not persisted: 

164 return 

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

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

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

168 if key not in overlayable: 

169 continue 

170 # Non-empty env var wins over config.toml (matches pydantic env_ignore_empty=True). 

171 if os.environ.get(f"{env_prefix}{key.upper()}", "") != "": 

172 continue 

173 # Legacy: set_setting used to persist None as "". Skip rather than 

174 # warn so a stale config doesn't spam logs on every CLI invocation. 

175 if raw == "": 

176 continue 

177 try: 

178 setattr(cfg, key, raw) 

179 except (ValueError, TypeError) as exc: 

180 log.warning( 

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

182 key, 

183 root, 

184 exc, 

185 )