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 

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 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.toml" 

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. 

74 """ 

75 path = _config_path(data_root) 

76 present = {k: v for k, v in sorted(settings.items()) if v is not None} 

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

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

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

80 

81 

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

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

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

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

86 

87 

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

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

90 with _config_write_lock(data_root): 

91 current = load(data_root) 

92 current[key] = value 

93 save(data_root, current) 

94 

95 

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

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

98 with _config_write_lock(data_root): 

99 current = load(data_root) 

100 current.pop(key, None) 

101 save(data_root, current) 

102 

103 

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

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

106 with _config_write_lock(data_root): 

107 current = load(data_root) 

108 current.update(updates) 

109 save(data_root, current) 

110 

111 

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

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

114 with _config_write_lock(data_root): 

115 current = load(data_root) 

116 for key in keys: 

117 current.pop(key, None) 

118 save(data_root, current) 

119 

120 

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

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

123 

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

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

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

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

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

129 """ 

130 with _config_write_lock(data_root): 

131 current = load(data_root) 

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

133 current[key] = new_value 

134 save(data_root, current) 

135 return result 

136 

137 

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

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

140 

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

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

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

144 

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

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

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

148 """ 

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

150 return 

151 log = logging.getLogger(__name__) 

152 try: 

153 persisted = load(root) 

154 except (OSError, ValueError): 

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

156 return 

157 if not persisted: 

158 return 

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

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

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

162 if key not in overlayable: 

163 continue 

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

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

166 continue 

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

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

169 if raw == "": 

170 continue 

171 try: 

172 setattr(cfg, key, raw) 

173 except (ValueError, TypeError) as exc: 

174 log.warning( 

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

176 key, 

177 root, 

178 exc, 

179 )