Coverage for src/lilbee/core/settings.py: 100%
242 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Persistent settings stored in config.toml alongside the data directory."""
3import logging
4import os
5import threading
6import tomllib
7import weakref
8from collections.abc import Callable, Generator, Iterable, Mapping
9from contextlib import contextmanager
10from dataclasses import dataclass, field, replace
11from pathlib import Path, PurePath
12from typing import Any, TypeVar
14import tomli_w
16from lilbee.config_meta import MODEL_ROLE_FIELDS, WRITABLE_CONFIG_FIELDS
17from lilbee.core.config import CONFIG_FILE_NAME, Config, cfg
18from lilbee.core.config.defaults import SKIP_TOML_ENV
19from lilbee.core.config.enums import SettingSource
20from lilbee.core.config.load_warnings import warn_on_load
21from lilbee.core.config.model import refusal_sentence, toml_value, toml_values
22from lilbee.core.config.resolve import (
23 PROFILE_NAME_KEY,
24 PROFILE_TABLE,
25 PROFILE_VALUES_KEY,
26 ROOT_DERIVED_FIELDS,
27 Resolved,
28 read_layers,
29 resolve_all,
30 sanitize_soft_fields,
31 toml_is_readable,
32)
33from lilbee.core.security import file_lock_or_warn, harden_private_file, write_private_text
35log = logging.getLogger(__name__)
37_settings_lock = threading.Lock()
39T = TypeVar("T")
41# A server, a CLI invocation, and an MCP process routinely run against the same
42# data root, so the in-process mutex alone lets two of them interleave a
43# read-modify-write and silently drop each other's keys.
44_CONFIG_LOCK_TIMEOUT_S = 10.0
46# An overlay leaves these on cfg when the root gives them no value of its own.
47HELD_AT_BUILT_IN = ROOT_DERIVED_FIELDS | set(MODEL_ROLE_FIELDS)
50def _config_path(data_root: Path) -> Path:
51 return data_root / CONFIG_FILE_NAME
54@contextmanager
55def _config_write_lock(data_root: Path) -> Generator[None, None, None]:
56 """Serialize a config read-modify-write across threads and processes.
58 A lock timeout falls through to the write rather than failing the caller:
59 losing a settings update to a stale lock file is worse than the interleave
60 the lock exists to prevent, which is already rare.
61 """
62 path = _config_path(data_root)
63 path.parent.mkdir(parents=True, exist_ok=True)
64 with _settings_lock, file_lock_or_warn(path, _CONFIG_LOCK_TIMEOUT_S):
65 yield
68def load(data_root: Path) -> dict[str, Any]:
69 """Read all settings from config.toml. Returns {} if file is missing.
71 Values keep the types TOML gave them. Stringifying here used to turn a
72 ``true`` into ``"True"`` in memory, which the next save then wrote back
73 quoted, so the file drifted away from valid types for its own fields.
75 A file that changes while it is read raises ConfigChangedWhileReadError: an editor
76 that rewrites the file in place can leave it half written at that moment, and a
77 read-modify-write of that half would drop every key it does not hold yet.
78 """
79 path = _config_path(data_root)
80 if not path.exists():
81 return {}
82 harden_private_file(path)
83 before = _stamp_of(path)
84 with path.open("rb") as f:
85 values = dict(tomllib.load(f))
86 if _stamp_of(path) != before:
87 raise ConfigChangedWhileReadError(f"{path} changed while it was read; try again")
88 return values
91class ConfigChangedWhileReadError(OSError):
92 """config.toml changed while it was read, so the reading can hold half of a write."""
95class ReadOnlyConfigError(PermissionError):
96 """config.toml exists and its owner has made it read-only; a write would replace it."""
99def save(data_root: Path, settings: dict[str, Any]) -> None:
100 """Write *settings* to config.toml.
102 ``tomli_w`` is the write half of the stdlib ``tomllib`` used by ``load``.
103 The emitter this replaced escaped strings by hand and stringified anything
104 that was not a bool or a number, so a list value was persisted as its
105 quoted repr and read back as text. A control character it escaped wrongly
106 was worse still: the reader discards the whole file on a parse error, so
107 one bad value silently wiped every other setting.
109 A ``None`` is dropped rather than written. TOML has no null, and the old
110 emitter wrote the literal string "None", which then read back as a set
111 value instead of an absent one. A path is written as its string: tomli_w
112 refuses ``Path`` objects, and the config's path fields hold them after
113 validation.
114 """
115 path = _config_path(data_root)
116 if path.exists() and not os.access(path, os.W_OK):
117 # The atomic replace swaps the file through a temp file, which a read-only file
118 # does not stop: refuse here, the one writer every surface goes through.
119 raise ReadOnlyConfigError(f"{path} is read-only; make it writable and try again")
120 present = {
121 k: str(v) if isinstance(v, PurePath) else v
122 for k, v in sorted(settings.items())
123 if v is not None
124 }
125 # config.toml can hold provider API keys, so it gets the same owner-only
126 # treatment as the session token rather than a post-hoc chmod.
127 write_private_text(path, tomli_w.dumps(present))
130def get(data_root: Path, key: str) -> str | None:
131 """Look up a single key from config.toml, as text for callers that want text."""
132 value = load(data_root).get(key)
133 return None if value is None else str(value)
136def set_value(data_root: Path, key: str, value: Any) -> None:
137 """Read-modify-write a single key in config.toml."""
138 with _config_write_lock(data_root):
139 current = load(data_root)
140 current[key] = value
141 save(data_root, current)
144def delete_value(data_root: Path, key: str) -> None:
145 """Remove a key from config.toml. No-op if key doesn't exist."""
146 with _config_write_lock(data_root):
147 current = load(data_root)
148 current.pop(key, None)
149 save(data_root, current)
152def update_values(data_root: Path, updates: dict[str, Any]) -> None:
153 """Batch update multiple keys in config.toml (single write)."""
154 with _config_write_lock(data_root):
155 current = load(data_root)
156 current.update(updates)
157 save(data_root, current)
160def delete_values(data_root: Path, keys: list[str]) -> None:
161 """Batch delete multiple keys from config.toml (single write, none when no key is present)."""
162 with _config_write_lock(data_root):
163 current = load(data_root)
164 present = [key for key in dict.fromkeys(keys) if key in current]
165 if not present:
166 return
167 for key in present:
168 del current[key]
169 save(data_root, current)
172def write_profile_table(
173 data_root: Path, name: str, values: Mapping[str, Any], drop: Iterable[str] = ()
174) -> None:
175 """Replace the ``[profile]`` table with *name* and *values* and remove the *drop* keys.
177 Other top-level keys stay as they are; one write does both.
178 """
179 table: dict[str, Any] = {PROFILE_NAME_KEY: name}
180 if values:
181 table[PROFILE_VALUES_KEY] = dict(values)
182 with _config_write_lock(data_root):
183 current = load(data_root)
184 for key in drop:
185 current.pop(key, None)
186 current[PROFILE_TABLE] = table
187 save(data_root, current)
190def mutate_value(data_root: Path, key: str, fn: Callable[[Any], tuple[Any, T]]) -> T:
191 """Read-modify-write a single key under the config lock, atomically.
193 ``fn`` receives the key's persisted value (or None if absent) read *inside*
194 the lock and returns ``(new_value, result)``; the new value is written and
195 the result is returned. Unlike a read-then-:func:`set_value`, the whole
196 compound update is serialized across threads and processes, so two callers
197 updating a dict-valued key cannot lose each other's change.
198 """
199 with _config_write_lock(data_root):
200 current = load(data_root)
201 new_value, result = fn(current.get(key))
202 current[key] = new_value
203 save(data_root, current)
204 return result
207def _assign_resolved(
208 config: Config, resolved: Mapping[str, Resolved], keys: Iterable[str], root: Path
209) -> list[str]:
210 """Set each of *keys* on *config* to its resolved value, warning on one the field rejects.
212 Returns the keys it set.
213 """
214 sanitized = sanitize_soft_fields(dict(resolved), root)
215 assigned: list[str] = []
216 for key in keys:
217 entry = sanitized[key]
218 try:
219 setattr(config, key, entry.value)
220 except (ValueError, TypeError):
221 # A bad variable is the entry point's to refuse, in its one line.
222 if entry.source is not SettingSource.ENV:
223 warn_on_load(
224 f"Ignoring {entry.source} value in {root}: {refusal_sentence(key, entry.value)}"
225 )
226 continue
227 assigned.append(key)
228 return assigned
231def sync_from_resolver(config: Config, keys: Iterable[str]) -> None:
232 """Set each of *keys* on *config* to the value the resolver gives under its data root."""
233 root = config.data_root
234 _assign_resolved(config, resolve_all(read_layers(root)), keys, root)
237def _overlay_file(root: Path) -> Path | None:
238 """The config.toml an overlay of *root* reads: None without one or under the skip switch."""
239 path = _config_path(root)
240 if os.environ.get(SKIP_TOML_ENV) == "1" or not path.exists():
241 return None
242 return path
245def overlay_stamp(root: Path) -> tuple[int, int, int] | None:
246 """What a write of the config.toml an overlay of *root* reads changes; None without one."""
247 path = _overlay_file(root)
248 return None if path is None else _stamp_of(path)
251def _stamp_of(path: Path) -> tuple[int, int, int] | None:
252 """What a write of the file at *path* changes; None when it is missing."""
253 try:
254 stat = path.stat()
255 except OSError:
256 return None
257 return stat.st_mtime_ns, stat.st_size, stat.st_ino
260@dataclass(frozen=True)
261class OverlayReading:
262 """What one overlay read from a root's config.toml: the file's stamp and the values it set."""
264 root: Path
265 stamp: tuple[int, int, int] | None
266 values: Mapping[str, Any]
269class _LastOverlay:
270 """The reading of the last overlay onto cfg, so a follower started later begins from it."""
272 def __init__(self) -> None:
273 self._reading: OverlayReading | None = None
274 self._lock = threading.Lock()
276 def record(self, reading: OverlayReading) -> None:
277 with self._lock:
278 self._reading = reading
280 def of(self, root: Path) -> OverlayReading | None:
281 with self._lock:
282 reading = self._reading
283 return reading if reading is not None and reading.root == root else None
286_last_overlay = _LastOverlay()
289def overlay_persisted_settings(root: Path) -> None:
290 """Set every overlayable key on cfg to its resolved value under ``root``.
292 A value in ``<root>/config.toml`` that Config refuses is left out with a warning
293 that names the file, and the setting keeps the value it had. A file that does not
294 parse is reported and changes nothing. A root-derived path at its built-in keeps the
295 value the caller set from ``root``, and a model role at its built-in keeps the
296 model cfg holds, so entering a fresh root leaves it usable. A root
297 without the file resolves as a root with an empty one, and
298 ``LILBEE_SKIP_TOML_CONFIG=1`` disables the overlay. The file is read once, under
299 the lock a writer takes, and ``last_overlay`` returns that reading.
300 """
301 stamp, values = read_persisted(root)
302 for key, value in (values or {}).items():
303 setattr(cfg, key, value)
304 _last_overlay.record(OverlayReading(root, stamp, values or {}))
307def last_overlay(root: Path) -> OverlayReading | None:
308 """The reading the last overlay took, when it was of *root*; None otherwise."""
309 return _last_overlay.of(root)
312def persisted_values(root: Path) -> dict[str, Any] | None:
313 """The value the overlay of *root* sets on each key, as cfg holds it.
315 A key the overlay leaves alone is absent: a value Config refuses, a root-derived
316 path at its built-in, and a model role at its built-in. None under the skip switch
317 or when the file does not parse, because the overlay then changes nothing.
318 """
319 if os.environ.get(SKIP_TOML_ENV) == "1":
320 return None
321 path = _config_path(root)
322 accepted: Mapping[str, Any] = {}
323 if path.exists():
324 harden_private_file(path)
325 accepted = toml_values(path)
326 if not toml_is_readable(path):
327 return None
328 layers = read_layers(root)
329 refused = (layers.user.keys() - accepted.keys()) | layers.refused_user
330 resolved = resolve_all(
331 replace(layers, user={key: layers.user[key] for key in layers.user.keys() - refused})
332 )
333 keys = [
334 key
335 for key in sorted((set(WRITABLE_CONFIG_FIELDS) | MODEL_ROLE_FIELDS) - refused)
336 if not (key in HELD_AT_BUILT_IN and resolved[key].source is SettingSource.BUILT_IN)
337 ]
338 trial = cfg.model_copy()
339 return {key: getattr(trial, key) for key in _assign_resolved(trial, resolved, keys, root)}
342def read_persisted(root: Path) -> tuple[tuple[int, int, int] | None, dict[str, Any] | None]:
343 """The stamp of *root*'s config.toml and ``persisted_values``, from one locked read.
345 The read holds the lock a writer takes, so it never sees a write by lilbee in
346 progress. Without the file there is nothing to lock, so the read creates no folder
347 or lock file. Another program can rewrite the file in place without the lock: when
348 the stamp moves during the read, the reading can hold half of that write, so the
349 values are None, as for a file that does not parse, and the stamp is None, so the
350 next check reads the file again.
351 """
352 if overlay_stamp(root) is None:
353 return _stable_reading(root)
354 with _config_write_lock(root):
355 return _stable_reading(root)
358def _stable_reading(
359 root: Path,
360) -> tuple[tuple[int, int, int] | None, dict[str, Any] | None]:
361 """``persisted_values`` of *root* and its stamp; both None when the stamp moved during it."""
362 stamp = overlay_stamp(root)
363 values = persisted_values(root)
364 if overlay_stamp(root) != stamp:
365 return None, None
366 return stamp, values
369@dataclass
370class _RegistryWatch:
371 """The config.toml state one Config's registry was read at, and how many blocks hold it.
373 ``lock`` guards the pin count and the swap of the stamp and the registry. It is never
374 held across file IO. ``reading`` is held by the one caller that reads the file; any
375 other caller uses the registry as it is.
376 """
378 stamp: tuple[int, int, int] | None = None
379 pins: int = 0
380 lock: threading.Lock = field(default_factory=threading.Lock)
381 reading: threading.Lock = field(default_factory=threading.Lock)
384class _RegistryWatches:
385 """One watch per live Config, keyed by identity and dropped with the Config."""
387 def __init__(self) -> None:
388 self._by_config: dict[int, _RegistryWatch] = {}
389 self._lock = threading.Lock()
391 def of(self, config: Config) -> _RegistryWatch:
392 with self._lock:
393 watch = self._by_config.get(id(config))
394 if watch is None:
395 watch = self._by_config[id(config)] = _RegistryWatch()
396 weakref.finalize(config, self._by_config.pop, id(config), None)
397 return watch
400_registry_watches = _RegistryWatches()
403def refresh_registry(config: Config) -> bool:
404 """Read the registry from config.toml again if another process changed the file.
406 Costs an existence check and one stat() when the file is unchanged. A pinned registry
407 is left alone. Returns False when it is pinned. A caller that finds another caller
408 reading the file does not wait: it uses the registry as it is. The registry and the
409 stamp change together under one small lock, inside the lock a writer takes.
410 """
411 watch = _registry_watches.of(config)
412 if watch.pins:
413 return False
414 if not watch.reading.acquire(blocking=False):
415 return True
416 try:
417 stamp = overlay_stamp(config.data_root)
418 if stamp == watch.stamp:
419 return True
420 pinned = False
422 def swap(roots: Any) -> None:
423 nonlocal pinned
424 with watch.lock:
425 if watch.pins:
426 pinned = True
427 return
428 if roots is not None:
429 config.linked_roots = roots
430 watch.stamp = stamp
432 _read_roots(config, swap)
433 return not pinned
434 finally:
435 watch.reading.release()
438@contextmanager
439def pin_registry(config: Config) -> Generator[None, None, None]:
440 """Hold the registry *config* has now: ``refresh_registry`` does nothing in the block."""
441 watch = _registry_watches.of(config)
442 with watch.lock:
443 watch.pins += 1
444 try:
445 yield
446 finally:
447 with watch.lock:
448 watch.pins -= 1
451@contextmanager
452def fresh_registry_pinned(config: Config) -> Generator[None, None, None]:
453 """Read the registry again if the file changed, then hold it for the block.
455 A block that reads the registry and then resolves keys against it sees one registry.
456 """
457 refresh_registry(config)
458 with pin_registry(config):
459 yield
462def _read_roots(config: Config, accept: Callable[[Any], None]) -> None:
463 """Pass *accept* the registry config.toml holds, None without one, under a writer's lock.
465 *accept* runs inside the lock, so a writer cannot change the file between the read
466 and the assignment.
467 """
468 path = _overlay_file(config.data_root)
469 if path is None:
470 accept(None)
471 return
472 with _config_write_lock(config.data_root):
473 accept(toml_value(path, "linked_roots"))
476def overlay_persisted_roots(config: Config) -> None:
477 """Set on *config* the source registry its config.toml holds, under the lock a writer takes.
479 A file without a registry sets nothing. A file that does not parse, or a
480 registry Config refuses, is reported and sets nothing.
481 ``LILBEE_SKIP_TOML_CONFIG=1`` disables it.
482 """
484 def assign(roots: Any) -> None:
485 if roots is not None:
486 config.linked_roots = roots
488 _read_roots(config, assign)