refactor(config): make load semantics explicit

.agent/design.md

@@ -27,3 +27,9 @@ A bugfix should make the protected invariant clear, change the smallest surface
 ## Explicit over magical
 
 Configuration must be declared explicitly in `config/schema.py` Pydantic models. Error handling should raise clear exceptions rather than silently correcting bad input. Provider auto-detection exists, but every resolution path must be traceable from the factory to the concrete provider class.
+
+## Configuration has an explicit owner
+
+`FileConfigRepository` owns config-file reads, validation, revisions, and atomic writes. Process entry points may use the explicit functions in `config/loader.py`; components with an explicit config path should keep their own repository instance instead of importing a mutable global config object.
+
+Persisted and runtime views are separate: `load_raw_config()` / `load_raw()` preserve `${VAR}` references for editing, while `load_effective_config()` / `load_effective()` return an isolated snapshot with references resolved. Read-modify-write flows must use `update()` / `update_config()` so a stale object cannot silently overwrite a newer change. Loading config is side-effect free; process-wide policies are applied explicitly during runtime startup.

.agent/gotchas.md

@@ -6,7 +6,7 @@
 
 ## Config `${VAR}` References
 
-`config/loader.py` resolves `${VAR}` patterns in `config.json` at load time. This is **not** a shell-like default-value syntax. If the environment variable is missing, `load_config` raises `ValueError` and the agent falls back to default configuration.
+`load_raw_config()` and `FileConfigRepository.load_raw()` preserve `${VAR}` patterns so Settings can safely edit and save the persisted representation. Runtime entry points use `load_effective_config()` / `load_effective()` to resolve them in an isolated snapshot. This is **not** a shell-like default-value syntax. If a referenced variable is missing, effective loading raises `ValueError`.
 
 Example valid usage:
 ```json

.agent/security.md

@@ -16,7 +16,7 @@ Shell execution (`ExecTool`, `agent/tools/shell.py`) also respects `restrict_to_
 
 All outbound HTTP requests from agent tools must pass through `validate_url_target` (`security/network.py`). By default it blocks loopback, RFC1918 private addresses, CGNAT ranges, link-local ranges, and cloud metadata endpoints (including `169.254.169.254`).
 
-The only escape hatch is `configure_ssrf_whitelist(cidrs)`, which reads from `config.tools.ssrf_whitelist` at load time.
+The only escape hatch is `configure_ssrf_whitelist(cidrs)`. Runtime entry points explicitly apply `config.tools.ssrf_whitelist` after loading the effective config; ordinary config reads must not mutate this process-wide policy.
 
 HTTP/SSE MCP transports are part of this boundary: validate configured MCP URLs before probing or constructing clients, and validate each outgoing HTTP request before redirects are followed. Local/private HTTP MCP endpoints are allowed only through the explicit SSRF whitelist. Stdio MCP servers are not part of the HTTP SSRF path.
 

docs/configuration.md

@@ -13,6 +13,46 @@ For setup and runtime failures, follow the diagnosis order in [`troubleshooting.
 > [!NOTE]
 > If your config file is older than the current schema, you can refresh it without overwriting your existing values: run `nanobot onboard`, then answer `N` when asked whether to overwrite the config. nanobot will merge in missing default fields and keep your current settings.
 
+## Python Configuration API
+
+> [!WARNING]
+> `nanobot.config.load_config` and `nanobot.config.loader.load_config` have been
+> removed. There is intentionally no compatibility alias because the old name
+> did not distinguish the persisted representation from the runtime view.
+
+Python embedders and plugins must choose the view they need explicitly:
+
+```python
+from pathlib import Path
+
+from nanobot.config import (
+    apply_config_runtime_policies,
+    load_effective_config,
+    load_raw_config,
+)
+
+path = Path.home() / ".nanobot" / "config.json"
+
+# For settings editors and persistence flows: preserves ${VAR} placeholders.
+persisted = load_raw_config(path)
+
+# For agents, providers, channels, and tools: resolves ${VAR} placeholders.
+runtime = load_effective_config(path)
+apply_config_runtime_policies(runtime)
+```
+
+Use this migration mapping:
+
+| Previous call | Replacement |
+|---|---|
+| `load_config(path)` used to inspect or edit persisted values | `load_raw_config(path)` |
+| `resolve_config_env_vars(load_config(path))` used at runtime | `load_effective_config(path)` |
+| Loading followed by implicit process-policy setup | `load_effective_config(path)`, then `apply_config_runtime_policies(config)` at the runtime boundary |
+
+This is a Python API breaking change for embedders and plugins that import the
+old function. The `config.json` format is unchanged, and the bundled CLI,
+gateway, and WebUI already use the explicit APIs.
+
 ## Configuration Guides
 
 This page is the complete configuration reference. For task-oriented setup, use

nanobot/agent/tools/mcp.py

@@ -1162,9 +1162,9 @@ async def reload_servers(state: Any, registry: ToolRegistry) -> dict[str, Any]:
                 "requires_restart": True,
             }
         try:
-            from nanobot.config.loader import load_config, resolve_config_env_vars
+            from nanobot.config.loader import load_effective_config
 
-            config = resolve_config_env_vars(load_config())
+            config = load_effective_config()
             next_servers = dict(config.tools.mcp_servers)
         except Exception as exc:
             logger.warning("MCP hot reload could not read config: {}", exc)

nanobot/agent/tools/web.py

@@ -315,8 +315,9 @@ class WebSearchTool(Tool):
         config_loader = None
         if ctx.provider_snapshot_loader is not None:
             def config_loader():
-                from nanobot.config.loader import load_config, resolve_config_env_vars
-                return resolve_config_env_vars(load_config()).tools.web.search
+                from nanobot.config.loader import load_effective_config
+
+                return load_effective_config().tools.web.search
         return cls(
             config=ctx.config.web.search,
             proxy=ctx.config.web.proxy,

nanobot/channels/base.py

@@ -52,9 +52,9 @@ class BaseChannel(ABC):
                 resolve_transcription_config,
                 transcribe_audio_file,
             )
-            from nanobot.config.loader import load_config
+            from nanobot.config.loader import load_raw_config
 
-            return await transcribe_audio_file(file_path, resolve_transcription_config(load_config()))
+            return await transcribe_audio_file(file_path, resolve_transcription_config(load_raw_config()))
         except Exception:
             self.logger.exception("Audio transcription failed")
             return ""

nanobot/channels/feishu.py

@@ -626,38 +626,44 @@ def sync_saved_feishu_identity_boundary(
     if not current_identity_key:
         return False
 
-    from nanobot.config.loader import load_config, save_config
-
-    full_config = load_config()
-    feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
-    if not isinstance(feishu_cfg, dict):
-        feishu_cfg = {}
+    from nanobot.config.loader import update_config
 
     defaults = FeishuChannel.default_config()
-    previous_identity_key = ""
-    for spec in feishu_instance_specs(feishu_cfg, defaults):
-        if spec.instance_id == instance_id:
-            previous_identity_key = str(
-                spec.config.get("identityKey") or spec.config.get("identity_key") or ""
-            )
-            break
+    access_cleared = False
 
-    access_cleared = bool(previous_identity_key and previous_identity_key != current_identity_key)
-    values: dict[str, Any] = {"identityKey": current_identity_key}
-    if access_cleared:
-        values["allowFrom"] = []
-        values["allow_from"] = []
-        clear_channel(runtime_channel_name("feishu", instance_id))
+    def mutate(full_config: Any) -> None:
+        nonlocal access_cleared
+        feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
+        if not isinstance(feishu_cfg, dict):
+            feishu_cfg = {}
 
-    if not previous_identity_key or access_cleared:
-        feishu_cfg = update_feishu_instance_preserving_shape(
-            feishu_cfg,
-            defaults,
-            instance_id,
-            values,
+        previous_identity_key = ""
+        for spec in feishu_instance_specs(feishu_cfg, defaults):
+            if spec.instance_id == instance_id:
+                previous_identity_key = str(
+                    spec.config.get("identityKey") or spec.config.get("identity_key") or ""
+                )
+                break
+
+        access_cleared = bool(
+            previous_identity_key and previous_identity_key != current_identity_key
         )
-        setattr(full_config.channels, "feishu", feishu_cfg)
-        save_config(full_config)
+        values: dict[str, Any] = {"identityKey": current_identity_key}
+        if access_cleared:
+            values["allowFrom"] = []
+            values["allow_from"] = []
+
+        if not previous_identity_key or access_cleared:
+            full_config.channels.feishu = update_feishu_instance_preserving_shape(
+                feishu_cfg,
+                defaults,
+                instance_id,
+                values,
+            )
+
+    update_config(mutate)
+    if access_cleared:
+        clear_channel(runtime_channel_name("feishu", instance_id))
 
     return access_cleared
 
@@ -669,19 +675,13 @@ def save_registration_result(
     name: str | None = None,
 ) -> None:
     """Persist a successful Feishu/Lark registration result to config.json."""
-    from nanobot.config.loader import load_config, save_config
+    from nanobot.config.loader import update_config
 
-    full_config = load_config()
-    feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
-    if not isinstance(feishu_cfg, dict):
-        feishu_cfg = {}
     defaults = FeishuChannel.default_config()
     app_id = str(result["app_id"]).strip()
     domain = str(result.get("domain", "feishu") or "feishu").strip().lower()
     domain = "lark" if domain == "lark" else "feishu"
-    previous_identity_key = _saved_feishu_instance_identity_key(feishu_cfg, defaults, instance_id)
     next_identity_key = _feishu_app_identity_key(app_id, domain)
-    identity_changed = bool(previous_identity_key and previous_identity_key != next_identity_key)
     identity: dict[str, str] = {}
     with suppress(Exception):
         identity = fetch_feishu_app_identity(
@@ -698,18 +698,35 @@ def save_registration_result(
         "enabled": True,
         **identity,
     }
+    identity_changed = False
+
+    def mutate(full_config: Any) -> None:
+        nonlocal identity_changed
+        feishu_cfg = getattr(full_config.channels, "feishu", None) or {}
+        if not isinstance(feishu_cfg, dict):
+            feishu_cfg = {}
+        previous_identity_key = _saved_feishu_instance_identity_key(
+            feishu_cfg,
+            defaults,
+            instance_id,
+        )
+        identity_changed = bool(
+            previous_identity_key and previous_identity_key != next_identity_key
+        )
+        next_values = dict(values)
+        if identity_changed:
+            next_values["allowFrom"] = []
+            next_values["allow_from"] = []
+        full_config.channels.feishu = upsert_feishu_instance(
+            feishu_cfg,
+            defaults,
+            instance_id,
+            next_values,
+        )
+
+    update_config(mutate)
     if identity_changed:
-        values["allowFrom"] = []
-        values["allow_from"] = []
         clear_channel(runtime_channel_name("feishu", instance_id))
-    feishu_cfg = upsert_feishu_instance(
-        feishu_cfg,
-        defaults,
-        instance_id,
-        values,
-    )
-    setattr(full_config.channels, "feishu", feishu_cfg)
-    save_config(full_config)
 
 
 def refresh_saved_feishu_identities(config: Any | None = None) -> bool:
@@ -723,13 +740,13 @@ def refresh_saved_feishu_identities(config: Any | None = None) -> bool:
     if not FEISHU_AVAILABLE:
         return False
 
-    from nanobot.config.loader import load_config, save_config
+    from nanobot.config.loader import load_raw_config, update_config
 
-    full_config = config or load_config()
-    feishu_cfg = getattr(full_config.channels, "feishu", None)
+    source_config = config or load_raw_config()
+    feishu_cfg = getattr(source_config.channels, "feishu", None)
     defaults = FeishuChannel.default_config()
     specs = feishu_instance_specs(feishu_cfg, defaults)
-    updated = False
+    fetched: dict[str, tuple[tuple[str, str, str], dict[str, str]]] = {}
 
     for spec in specs:
         instance = spec.config
@@ -753,20 +770,50 @@ def refresh_saved_feishu_identities(config: Any | None = None) -> bool:
         if not identity:
             identity = {"identityFetchedAt": _identity_timestamp()}
 
-        feishu_cfg = update_feishu_instance_preserving_shape(
-            feishu_cfg,
-            defaults,
-            spec.instance_id,
+        fetched[spec.instance_id] = (
+            (app_id, app_secret, str(instance.get("domain") or "feishu")),
             identity,
         )
-        updated = True
 
-    if not updated:
+    if not fetched:
         return False
 
-    setattr(full_config.channels, "feishu", feishu_cfg)
-    save_config(full_config)
-    return True
+    updated = False
+
+    def mutate(full_config: Any) -> None:
+        nonlocal updated
+        current_cfg = getattr(full_config.channels, "feishu", None)
+        for spec in feishu_instance_specs(current_cfg, defaults):
+            fetched_entry = fetched.get(spec.instance_id)
+            if fetched_entry is None:
+                continue
+            expected_credentials, identity = fetched_entry
+            instance = spec.config
+            current_credentials = (
+                str(instance.get("appId") or instance.get("app_id") or "").strip(),
+                str(instance.get("appSecret") or instance.get("app_secret") or "").strip(),
+                str(instance.get("domain") or "feishu"),
+            )
+            if current_credentials != expected_credentials:
+                continue
+            if (
+                instance.get("displayName")
+                or instance.get("avatarUrl")
+                or instance.get("identityFetchedAt")
+            ):
+                continue
+            current_cfg = update_feishu_instance_preserving_shape(
+                current_cfg,
+                defaults,
+                spec.instance_id,
+                identity,
+            )
+            updated = True
+        if updated:
+            full_config.channels.feishu = current_cfg
+
+    update_config(mutate)
+    return updated
 
 
 def qr_register(

nanobot/channels/manager.py

@@ -364,9 +364,9 @@ class ChannelManager:
                 "message": "WebSocket hosts the WebUI and is applied on restart.",
             }
 
-        from nanobot.config.loader import load_config
+        from nanobot.config.loader import load_effective_config
 
-        self.config = load_config()
+        self.config = load_effective_config()
         section = self._channel_section(name)
         if action == "disable":
             runtime_names = [name if not instance_id else f"{name}.{instance_id}"]

nanobot/cli/commands.py

@@ -640,7 +640,7 @@ def onboard(
     non_interactive_refresh: bool = typer.Option(False, "--refresh", help="Refresh config, preserving existing settings without prompting"),
 ):
     """Initialize nanobot configuration and workspace."""
-    from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
+    from nanobot.config.loader import get_config_path, load_raw_config, save_config, set_config_path
     from nanobot.config.schema import Config
 
     if config:
@@ -658,7 +658,7 @@ def onboard(
     # Create or update config
     if config_path.exists():
         if wizard:
-            config = _apply_workspace_override(load_config(config_path))
+            config = _apply_workspace_override(load_raw_config(config_path))
         else:
             should_refresh = non_interactive_refresh
             if not non_interactive_refresh:
@@ -677,7 +677,7 @@ def onboard(
                     should_refresh = True
 
             if should_refresh:
-                config = _apply_workspace_override(load_config(config_path))
+                config = _apply_workspace_override(load_raw_config(config_path))
                 save_config(config, config_path)
                 console.print(
                     f"[green]✓[/green] Config refreshed at {config_path} (existing values preserved)"
@@ -738,27 +738,25 @@ def onboard(
 
 def _onboard_plugins(config_path: Path) -> None:
     """Inject default config for all discovered channels (built-in + plugins)."""
-    import json
-
     from nanobot.channels.registry import discover_all
-    from nanobot.config.loader import merge_missing_defaults
+    from nanobot.config.loader import get_config_repository, merge_missing_defaults
 
     all_channels = discover_all()
     if not all_channels:
         return
 
-    with open(config_path, encoding="utf-8") as f:
-        data = json.load(f)
+    def mutate(config: Config) -> None:
+        for name, cls in all_channels.items():
+            existing = getattr(config.channels, name, None)
+            if not isinstance(existing, dict):
+                existing = {}
+            setattr(
+                config.channels,
+                name,
+                merge_missing_defaults(existing, cls.default_config()),
+            )
 
-    channels = data.setdefault("channels", {})
-    for name, cls in all_channels.items():
-        if name not in channels:
-            channels[name] = cls.default_config()
-        else:
-            channels[name] = merge_missing_defaults(channels[name], cls.default_config())
-
-    with open(config_path, "w", encoding="utf-8") as f:
-        json.dump(data, f, indent=2, ensure_ascii=False)
+    get_config_repository(config_path).update(mutate)
 
 
 def _print_enable_options(
@@ -798,7 +796,11 @@ def _model_display(config: Config) -> tuple[str, str]:
 
 def _load_runtime_config(config: str | None = None, workspace: str | None = None) -> Config:
     """Load config and optionally override the active workspace."""
-    from nanobot.config.loader import load_config, resolve_config_env_vars, set_config_path
+    from nanobot.config.loader import (
+        apply_config_runtime_policies,
+        load_effective_config,
+        set_config_path,
+    )
 
     config_path = None
     if config:
@@ -810,13 +812,14 @@ def _load_runtime_config(config: str | None = None, workspace: str | None = None
         console.print(f"[dim]Using config: {config_path}[/dim]")
 
     try:
-        loaded = resolve_config_env_vars(load_config(config_path))
+        loaded = load_effective_config(config_path)
     except ValueError as e:
         console.print(f"[red]Error: {e}[/red]")
         raise typer.Exit(1)
     _warn_deprecated_config_keys(config_path)
     if workspace:
         loaded.agents.defaults.workspace = workspace
+    apply_config_runtime_policies(loaded)
     return loaded
 
 
@@ -858,7 +861,7 @@ def _load_inspection_config(
     workspace: str | None = None,
 ) -> tuple[Path, Config]:
     """Load config for diagnostic commands without resolving secret env refs."""
-    from nanobot.config.loader import get_config_path, load_config, set_config_path
+    from nanobot.config.loader import get_config_path, load_raw_config, set_config_path
 
     config_path = None
     if config:
@@ -868,7 +871,7 @@ def _load_inspection_config(
 
     display_path = config_path or get_config_path()
     try:
-        loaded = load_config(config_path)
+        loaded = load_raw_config(config_path)
     except ValueError as exc:
         console.print(f"[red]Error: {exc}[/red]")
         raise typer.Exit(1) from exc
@@ -920,10 +923,10 @@ def _resolve_webui_config_path(config: str | None) -> Path:
 
 def _load_webui_setup_config(config_path: Path) -> Config:
     """Load config for first-run mutation without resolving env-var placeholders."""
-    from nanobot.config.loader import load_config
+    from nanobot.config.loader import load_raw_config
 
     try:
-        return load_config(config_path)
+        return load_raw_config(config_path)
     except ValueError as e:
         console.print(f"[red]Error: {e}[/red]")
         raise typer.Exit(1) from e
@@ -931,7 +934,7 @@ def _load_webui_setup_config(config_path: Path) -> Config:
 
 def _provider_setup_error(config: Config) -> str | None:
     """Return the provider setup error, or None when the current model can start."""
-    from nanobot.config.loader import resolve_config_env_vars
+    from nanobot.config.repository import resolve_config_env_vars
     from nanobot.providers.factory import build_provider_snapshot
 
     try:
@@ -2517,7 +2520,7 @@ def plugins_list(
 ):
     """List optional nanobot features."""
     from nanobot.channels.registry import discover_channel_names, discover_plugins
-    from nanobot.config.loader import load_config, set_config_path
+    from nanobot.config.loader import load_raw_config, set_config_path
 
     resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
     if resolved_config_path is not None:
@@ -2527,7 +2530,7 @@ def plugins_list(
         feature_support.optional_dependency_groups(),
         set(discover_channel_names()),
         discover_plugins(),
-        load_config(resolved_config_path),
+        load_raw_config(resolved_config_path),
     )
 
 
@@ -2688,19 +2691,21 @@ def _set_oauth_provider_as_main(
     config_path: str | None = None,
 ) -> None:
     """Persist an OAuth provider as the active agent provider."""
-    from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
+    from nanobot.config.loader import get_config_path, set_config_path, update_config
 
     resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
     if resolved_config_path is not None:
         set_config_path(resolved_config_path)
         console.print(f"[dim]Using config: {resolved_config_path}[/dim]")
 
-    config = load_config(resolved_config_path)
     selected_model = (model or "").strip() or _OAUTH_PROVIDER_DEFAULT_MODELS[provider_name]
-    config.agents.defaults.model_preset = None
-    config.agents.defaults.provider = provider_name
-    config.agents.defaults.model = selected_model
-    save_config(config, resolved_config_path)
+
+    def mutate(config: Config) -> None:
+        config.agents.defaults.model_preset = None
+        config.agents.defaults.provider = provider_name
+        config.agents.defaults.model = selected_model
+
+    update_config(mutate, resolved_config_path)
 
     saved_path = resolved_config_path or get_config_path()
     console.print(
@@ -2762,11 +2767,11 @@ def _login_openai_codex() -> None:
     try:
         from oauth_cli_kit import get_token, login_oauth_interactive
 
-        from nanobot.config.loader import load_config, resolve_config_env_vars
+        from nanobot.config.loader import load_effective_config
 
         proxy = None
         try:
-            proxy = resolve_config_env_vars(load_config()).providers.openai_codex.proxy or None
+            proxy = load_effective_config().providers.openai_codex.proxy or None
         except ValueError as e:
             console.print(f"[red]{e}[/red]")
             raise typer.Exit(1) from e

nanobot/cli/onboard.py

@@ -22,7 +22,7 @@ from nanobot.cli.models import (
     get_model_context_limit,
     get_model_suggestions,
 )
-from nanobot.config.loader import get_config_path, load_config
+from nanobot.config.loader import get_config_path, load_raw_config
 from nanobot.config.schema import Config, ModelPresetConfig
 
 console = Console()
@@ -1940,7 +1940,7 @@ def run_onboard(initial_config: Config | None = None) -> OnboardResult:
     else:
         config_path = get_config_path()
         if config_path.exists():
-            base_config = load_config()
+            base_config = load_raw_config()
         else:
             base_config = Config()
 

nanobot/config/__init__.py

@@ -1,6 +1,13 @@
 """Configuration module for nanobot."""
 
-from nanobot.config.loader import get_config_path, load_config
+from nanobot.config.loader import (
+    apply_config_runtime_policies,
+    get_config_path,
+    get_config_repository,
+    load_effective_config,
+    load_raw_config,
+    update_config,
+)
 from nanobot.config.paths import (
     get_cli_history_path,
     get_cron_dir,
@@ -13,12 +20,28 @@ from nanobot.config.paths import (
     get_workspace_path,
     is_default_workspace,
 )
+from nanobot.config.repository import (
+    ConfigCommit,
+    ConfigConflictError,
+    EffectiveConfigSnapshot,
+    FileConfigRepository,
+    PersistedConfigSnapshot,
+)
 from nanobot.config.schema import Config
 
 __all__ = [
     "Config",
-    "load_config",
+    "ConfigCommit",
+    "ConfigConflictError",
+    "EffectiveConfigSnapshot",
+    "FileConfigRepository",
+    "PersistedConfigSnapshot",
+    "apply_config_runtime_policies",
+    "load_effective_config",
+    "load_raw_config",
+    "update_config",
     "get_config_path",
+    "get_config_repository",
     "get_data_dir",
     "get_runtime_subdir",
     "get_media_dir",

nanobot/config/loader.py

@@ -1,92 +1,76 @@
-"""Configuration loading utilities."""
+"""Configuration loading and persistence entry points."""
 
-import json
-import os
-import re
+from __future__ import annotations
+
+from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 
-import pydantic
-from loguru import logger
-from pydantic import BaseModel
+from nanobot.config.repository import (
+    ConfigCommit,
+    FileConfigRepository,
+)
+from nanobot.config.schema import Config
 
-from nanobot.config.schema import Config, _resolve_tool_config_refs
-
-# Global variable to store current config path (for multi-instance support)
+# Default path for process entry points that do not receive an explicit path.
 _current_config_path: Path | None = None
-_schema_refs_ready = False
 
 
 def set_config_path(path: Path) -> None:
-    """Set the current config path (used to derive data directory)."""
+    """Set the default configuration path for subsequent entry-point calls."""
     global _current_config_path
     _current_config_path = path
 
 
 def get_config_path() -> Path:
-    """Get the configuration file path."""
+    """Get the default configuration file path."""
     if _current_config_path:
         return _current_config_path
     return Path.home() / ".nanobot" / "config.json"
 
 
-def load_config(config_path: Path | None = None) -> Config:
-    """
-    Load configuration from file or create default.
-
-    Args:
-        config_path: Optional path to config file. Uses default if not provided.
-
-    Returns:
-        Loaded configuration object.
-    """
-    global _schema_refs_ready
-    if not _schema_refs_ready:
-        _resolve_tool_config_refs()
-        _schema_refs_ready = True
-
-    path = config_path or get_config_path()
-
-    config = Config()
-    if path.exists():
-        try:
-            with open(path, encoding="utf-8") as f:
-                data = json.load(f)
-            data = _migrate_config(data)
-            config = Config.model_validate(data)
-        except (json.JSONDecodeError, ValueError, pydantic.ValidationError) as e:
-            raise ValueError(f"Failed to load config from {path}: {e}") from e
-
-    _apply_ssrf_whitelist(config)
-    return config
+def get_config_repository(config_path: Path | None = None) -> FileConfigRepository:
+    """Return an instance-scoped repository for *config_path*."""
+    return FileConfigRepository(config_path or get_config_path())
 
 
-def _apply_ssrf_whitelist(config: Config) -> None:
-    """Apply SSRF whitelist from config to the network security module."""
-    from nanobot.security.network import configure_ssrf_whitelist
+def load_raw_config(config_path: Path | None = None) -> Config:
+    """Load raw persisted config without applying process runtime policy."""
+    return get_config_repository(config_path).load_raw().config
 
-    configure_ssrf_whitelist(config.tools.ssrf_whitelist)
+
+def load_effective_config(config_path: Path | None = None) -> Config:
+    """Load a fresh runtime config with environment references resolved."""
+    return get_config_repository(config_path).load_effective().config
 
 
 def save_config(config: Config, config_path: Path | None = None) -> None:
+    """Atomically save a complete config.
+
+    New read-modify-write flows should use :func:`update_config` so concurrent
+    writers for the same file cannot silently overwrite one another.
     """
-    Save configuration to file.
+    get_config_repository(config_path).save(config)
 
-    Args:
-        config: Configuration to save.
-        config_path: Optional path to save to. Uses default if not provided.
-    """
-    path = config_path or get_config_path()
-    path.parent.mkdir(parents=True, exist_ok=True)
 
-    data = config.model_dump(mode="json", by_alias=True)
-    if config.providers.openai_codex.proxy is not None:
-        data.setdefault("providers", {})["openaiCodex"] = {
-            "proxy": config.providers.openai_codex.proxy,
-        }
+def update_config(
+    mutator: Callable[[Config], None],
+    config_path: Path | None = None,
+    *,
+    expected_revision: str | None = None,
+) -> ConfigCommit:
+    """Atomically mutate the latest raw config and return the resulting commit."""
+    return get_config_repository(config_path).update(
+        mutator,
+        expected_revision=expected_revision,
+    )
 
-    with open(path, "w", encoding="utf-8") as f:
-        json.dump(data, f, indent=2, ensure_ascii=False)
+
+def apply_config_runtime_policies(config: Config) -> None:
+    """Apply process-level policies when starting or reconfiguring a runtime."""
+    from nanobot.security.network import configure_ssrf_whitelist
+
+    configure_ssrf_whitelist(config.tools.ssrf_whitelist)
 
 
 def merge_missing_defaults(existing: Any, defaults: Any) -> Any:
@@ -103,109 +87,14 @@ def merge_missing_defaults(existing: Any, defaults: Any) -> Any:
     return merged
 
 
-_ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
-
-
-def resolve_config_env_vars(config: Config) -> Config:
-    """Return *config* with ``${VAR}`` env-var references resolved.
-
-    Walks in place so fields declared with ``exclude=True`` survive;
-    returns the same instance when no references are present.
-    Raises ``ValueError`` if a referenced variable is not set.
-    """
-    return _resolve_in_place(config)
-
-
-def _resolve_in_place(obj: Any) -> Any:
-    if isinstance(obj, str):
-        new = _ENV_REF_PATTERN.sub(_env_replace, obj)
-        return new if new != obj else obj
-    if isinstance(obj, BaseModel):
-        updates: dict[str, Any] = {}
-        for name in type(obj).model_fields:
-            old = getattr(obj, name)
-            new = _resolve_in_place(old)
-            if new is not old:
-                updates[name] = new
-        extras = obj.__pydantic_extra__
-        new_extras: dict[str, Any] | None = None
-        if extras:
-            resolved = {k: _resolve_in_place(v) for k, v in extras.items()}
-            if any(resolved[k] is not extras[k] for k in extras):
-                new_extras = resolved
-        if not updates and new_extras is None:
-            return obj
-        copy = obj.model_copy(update=updates) if updates else obj.model_copy()
-        if new_extras is not None:
-            copy.__pydantic_extra__ = new_extras
-        return copy
-    if isinstance(obj, dict):
-        resolved = {k: _resolve_in_place(v) for k, v in obj.items()}
-        return resolved if any(resolved[k] is not obj[k] for k in obj) else obj
-    if isinstance(obj, list):
-        resolved = [_resolve_in_place(v) for v in obj]
-        return resolved if any(nv is not ov for nv, ov in zip(resolved, obj)) else obj
-    return obj
-
-
-def _resolve_env_vars(obj: object) -> object:
-    """Recursively resolve ``${VAR}`` patterns in plain strings/dicts/lists."""
-    if isinstance(obj, str):
-        return _ENV_REF_PATTERN.sub(_env_replace, obj)
-    if isinstance(obj, dict):
-        return {k: _resolve_env_vars(v) for k, v in obj.items()}
-    if isinstance(obj, list):
-        return [_resolve_env_vars(v) for v in obj]
-    return obj
-
-
-def _env_replace(match: re.Match[str]) -> str:
-    name = match.group(1)
-    value = os.environ.get(name)
-    if value is None:
-        raise ValueError(
-            f"Environment variable '{name}' referenced in config is not set"
-        )
-    return value
-
-
-def _migrate_config(data: dict) -> dict:
-    """Migrate old config formats to current."""
-    agents = data.get("agents", {})
-    defaults = agents.get("defaults", {}) if isinstance(agents, dict) else {}
-    if isinstance(defaults, dict):
-        had_legacy_max_messages = (
-            "maxMessages" in defaults or "max_messages" in defaults
-        )
-        defaults.pop("maxMessages", None)
-        defaults.pop("max_messages", None)
-        if had_legacy_max_messages:
-            # TODO(next version): Remove this legacy cleanup branch; the schema
-            # will silently ignore this field once the warning grace period ends.
-            logger.warning(
-                "agents.defaults.maxMessages/max_messages is legacy and ignored; "
-                "replay max messages is now an internal safety cap. Remove it from "
-                "config. This compatibility warning will be removed in the next version."
-            )
-
-    # Move tools.exec.restrictToWorkspace → tools.restrictToWorkspace
-    tools = data.get("tools", {})
-    exec_cfg = tools.get("exec", {})
-    if "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
-        tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
-
-    # Move tools.myEnabled / tools.mySet → tools.my.{enable, allowSet}.
-    # The old flat keys shipped in the initial MyTool landing; wrapping them in a
-    # sub-config keeps `web` / `exec` / `my` symmetric and gives room to grow.
-    if "myEnabled" in tools or "mySet" in tools:
-        my_cfg = tools.setdefault("my", {})
-        if "myEnabled" in tools and "enable" not in my_cfg:
-            my_cfg["enable"] = tools.pop("myEnabled")
-        else:
-            tools.pop("myEnabled", None)
-        if "mySet" in tools and "allowSet" not in my_cfg:
-            my_cfg["allowSet"] = tools.pop("mySet")
-        else:
-            tools.pop("mySet", None)
-
-    return data
+__all__ = [
+    "apply_config_runtime_policies",
+    "get_config_path",
+    "get_config_repository",
+    "load_raw_config",
+    "load_effective_config",
+    "merge_missing_defaults",
+    "save_config",
+    "set_config_path",
+    "update_config",
+]

nanobot/config/repository.py

@@ -0,0 +1,299 @@
+"""Instance-scoped configuration persistence and resolution."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import stat
+import threading
+import uuid
+from collections.abc import Callable
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from loguru import logger
+from pydantic import BaseModel
+
+from nanobot.config.schema import Config
+
+ConfigMutator = Callable[[Config], None]
+
+_ENV_REF_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
+
+# Several WebUI config actions run in worker threads via asyncio.to_thread().
+# Repository instances for the same path therefore share a process-local lock.
+_path_locks_guard = threading.Lock()
+_path_locks: dict[Path, threading.RLock] = {}
+
+
+class ConfigConflictError(RuntimeError):
+    """Raised when a caller tries to update a stale configuration snapshot."""
+
+
+@dataclass(frozen=True)
+class PersistedConfigSnapshot:
+    """Validated configuration as represented by the persisted source."""
+
+    config: Config
+    path: Path
+    revision: str
+
+
+@dataclass(frozen=True)
+class EffectiveConfigSnapshot:
+    """Runtime configuration with environment references resolved."""
+
+    config: Config
+    path: Path
+    revision: str
+
+
+@dataclass(frozen=True)
+class ConfigCommit:
+    """One atomic configuration update and its semantic field changes."""
+
+    before: PersistedConfigSnapshot
+    after: PersistedConfigSnapshot
+    changed_paths: frozenset[str]
+
+
+def _lock_for_path(path: Path) -> threading.RLock:
+    key = path.expanduser().resolve(strict=False)
+    with _path_locks_guard:
+        return _path_locks.setdefault(key, threading.RLock())
+
+
+def _revision_for_bytes(raw: bytes | None) -> str:
+    if raw is None:
+        return "missing"
+    return f"sha256:{hashlib.sha256(raw).hexdigest()}"
+
+
+def _config_data(config: Config) -> dict[str, Any]:
+    data = config.model_dump(mode="json", by_alias=True)
+    if config.providers.openai_codex.proxy is not None:
+        data.setdefault("providers", {})["openaiCodex"] = {
+            "proxy": config.providers.openai_codex.proxy,
+        }
+    return data
+
+
+def _validate_config_data(data: dict[str, Any], path: Path) -> Config:
+    try:
+        return Config.model_validate(data)
+    except ValueError as exc:
+        raise ValueError(f"Failed to load config from {path}: {exc}") from exc
+
+
+def _read_snapshot(path: Path) -> PersistedConfigSnapshot:
+    if not path.exists():
+        return PersistedConfigSnapshot(Config(), path, "missing")
+
+    try:
+        raw = path.read_bytes()
+        data = json.loads(raw.decode("utf-8"))
+        if not isinstance(data, dict):
+            raise ValueError("config root must be a JSON object")
+        data = _migrate_config(data)
+        config = Config.model_validate(data)
+    except (UnicodeDecodeError, ValueError) as exc:
+        raise ValueError(f"Failed to load config from {path}: {exc}") from exc
+    return PersistedConfigSnapshot(config, path, _revision_for_bytes(raw))
+
+
+def _write_config_atomic(path: Path, data: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = json.dumps(data, indent=2, ensure_ascii=False)
+    tmp = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
+    existing_mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else None
+    try:
+        with open(tmp, "w", encoding="utf-8") as handle:
+            if existing_mode is not None:
+                try:
+                    os.chmod(tmp, existing_mode)
+                except OSError:
+                    if os.name != "nt":
+                        raise
+            handle.write(content)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(tmp, path)
+        with suppress(OSError, NotImplementedError):
+            directory_fd = os.open(path.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+    finally:
+        tmp.unlink(missing_ok=True)
+
+
+def _changed_paths(before: Any, after: Any, prefix: str = "") -> set[str]:
+    if isinstance(before, dict) and isinstance(after, dict):
+        changed: set[str] = set()
+        for key in before.keys() | after.keys():
+            path = f"{prefix}.{key}" if prefix else str(key)
+            if key not in before or key not in after:
+                changed.add(path)
+            else:
+                changed.update(_changed_paths(before[key], after[key], path))
+        return changed
+    if before != after:
+        return {prefix} if prefix else {"<root>"}
+    return set()
+
+
+class FileConfigRepository:
+    """Read and atomically update one configuration file.
+
+    The repository does not cache. Every read returns a new validated snapshot.
+    """
+
+    def __init__(self, path: str | Path):
+        self.path = Path(path).expanduser().resolve(strict=False)
+        self._lock = _lock_for_path(self.path)
+
+    def load_raw(self) -> PersistedConfigSnapshot:
+        """Load the persisted form without resolving secret references."""
+        with self._lock:
+            return _read_snapshot(self.path)
+
+    def load_effective(self) -> EffectiveConfigSnapshot:
+        """Load an isolated runtime snapshot with ``${VAR}`` references resolved."""
+        raw = self.load_raw()
+        effective = resolve_config_env_vars(raw.config.model_copy(deep=True))
+        return EffectiveConfigSnapshot(effective, raw.path, raw.revision)
+
+    def save(
+        self,
+        config: Config,
+        *,
+        expected_revision: str | None = None,
+    ) -> PersistedConfigSnapshot:
+        """Atomically save a complete config, optionally rejecting stale writes."""
+        with self._lock:
+            current = _read_snapshot(self.path)
+            if expected_revision is not None and current.revision != expected_revision:
+                raise ConfigConflictError(
+                    f"Config changed since revision {expected_revision}; "
+                    f"current revision is {current.revision}"
+                )
+            data = _config_data(config)
+            _validate_config_data(data, self.path)
+            _write_config_atomic(self.path, data)
+            return _read_snapshot(self.path)
+
+    def update(
+        self,
+        mutator: ConfigMutator,
+        *,
+        expected_revision: str | None = None,
+    ) -> ConfigCommit:
+        """Atomically apply a mutation to the latest persisted config."""
+        with self._lock:
+            before = _read_snapshot(self.path)
+            if expected_revision is not None and before.revision != expected_revision:
+                raise ConfigConflictError(
+                    f"Config changed since revision {expected_revision}; "
+                    f"current revision is {before.revision}"
+                )
+
+            before_data = _config_data(before.config)
+            draft = before.config.model_copy(deep=True)
+            mutator(draft)
+            after_data = _config_data(draft)
+            changed = frozenset(_changed_paths(before_data, after_data))
+            if not changed:
+                return ConfigCommit(before, before, changed)
+
+            _validate_config_data(after_data, self.path)
+            _write_config_atomic(self.path, after_data)
+            after = _read_snapshot(self.path)
+            return ConfigCommit(before, after, changed)
+
+
+def resolve_config_env_vars(config: Config) -> Config:
+    """Return *config* with ``${VAR}`` environment references resolved."""
+    return _resolve_in_place(config)
+
+
+def _resolve_in_place(obj: Any) -> Any:
+    if isinstance(obj, str):
+        new = _ENV_REF_PATTERN.sub(_env_replace, obj)
+        return new if new != obj else obj
+    if isinstance(obj, BaseModel):
+        updates: dict[str, Any] = {}
+        for name in type(obj).model_fields:
+            old = getattr(obj, name)
+            new = _resolve_in_place(old)
+            if new is not old:
+                updates[name] = new
+        extras = obj.__pydantic_extra__
+        new_extras: dict[str, Any] | None = None
+        if extras:
+            resolved = {key: _resolve_in_place(value) for key, value in extras.items()}
+            if any(resolved[key] is not extras[key] for key in extras):
+                new_extras = resolved
+        if not updates and new_extras is None:
+            return obj
+        copy = obj.model_copy(update=updates) if updates else obj.model_copy()
+        if new_extras is not None:
+            copy.__pydantic_extra__ = new_extras
+        return copy
+    if isinstance(obj, dict):
+        resolved = {key: _resolve_in_place(value) for key, value in obj.items()}
+        return resolved if any(resolved[key] is not obj[key] for key in obj) else obj
+    if isinstance(obj, list):
+        resolved = [_resolve_in_place(value) for value in obj]
+        return resolved if any(new is not old for new, old in zip(resolved, obj)) else obj
+    return obj
+
+
+def _env_replace(match: re.Match[str]) -> str:
+    name = match.group(1)
+    value = os.environ.get(name)
+    if value is None:
+        raise ValueError(f"Environment variable '{name}' referenced in config is not set")
+    return value
+
+
+def _migrate_config(data: dict[str, Any]) -> dict[str, Any]:
+    """Migrate old config formats to the current schema."""
+    agents = data.get("agents", {})
+    defaults = agents.get("defaults", {}) if isinstance(agents, dict) else {}
+    if isinstance(defaults, dict):
+        had_legacy_max_messages = "maxMessages" in defaults or "max_messages" in defaults
+        defaults.pop("maxMessages", None)
+        defaults.pop("max_messages", None)
+        if had_legacy_max_messages:
+            logger.warning(
+                "agents.defaults.maxMessages/max_messages is legacy and ignored; "
+                "replay max messages is now an internal safety cap. Remove it from "
+                "config. This compatibility warning will be removed in the next version."
+            )
+
+    tools = data.get("tools", {})
+    if not isinstance(tools, dict):
+        return data
+    exec_cfg = tools.get("exec", {})
+    if isinstance(exec_cfg, dict) and "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
+        tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
+
+    if "myEnabled" in tools or "mySet" in tools:
+        my_cfg = tools.setdefault("my", {})
+        if not isinstance(my_cfg, dict):
+            return data
+        if "myEnabled" in tools and "enable" not in my_cfg:
+            my_cfg["enable"] = tools.pop("myEnabled")
+        else:
+            tools.pop("myEnabled", None)
+        if "mySet" in tools and "allowSet" not in my_cfg:
+            my_cfg["allowSet"] = tools.pop("mySet")
+        else:
+            tools.pop("mySet", None)
+
+    return data

nanobot/nanobot.py

@@ -95,7 +95,10 @@ class Nanobot:
             model: Override the instance default model.
             model_preset: Override the instance default model preset.
         """
-        from nanobot.config.loader import load_config, resolve_config_env_vars
+        from nanobot.config.loader import (
+            apply_config_runtime_policies,
+            load_effective_config,
+        )
 
         ensure_single_model_selector(model=model, model_preset=model_preset)
         resolved: Path | None = None
@@ -104,7 +107,7 @@ class Nanobot:
             if not resolved.exists():
                 raise FileNotFoundError(f"Config not found: {resolved}")
 
-        config: Config = resolve_config_env_vars(load_config(resolved))
+        config: Config = load_effective_config(resolved)
         if workspace is not None:
             config.agents.defaults.workspace = str(
                 Path(workspace).expanduser().resolve()
@@ -116,6 +119,8 @@ class Nanobot:
         elif model_preset is not None:
             config.agents.defaults.model_preset = model_preset
 
+        apply_config_runtime_policies(config)
+
         loop = AgentLoop.from_config(
             config,
             image_generation_provider_configs=image_gen_provider_configs(config),

nanobot/optional_features.py

@@ -26,7 +26,7 @@ from nanobot.channels._setup import (
     stringify_channel_value,
 )
 from nanobot.channels.registry import DEFAULT_ENABLED_CHANNELS
-from nanobot.config.loader import merge_missing_defaults
+from nanobot.config.loader import get_config_repository, merge_missing_defaults
 from nanobot.config.schema import Config
 
 
@@ -266,29 +266,16 @@ def install_extra(
     return InstallResult(False, label, pip_cmd, failed_cmd=failed_cmd, output=output)
 
 
-def read_config_data(path: Path) -> dict[str, Any]:
-    if not path.exists():
-        return {}
-    with open(path, encoding="utf-8") as f:
-        return json.load(f)
-
-
-def write_config_data(path: Path, data: dict[str, Any]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w", encoding="utf-8") as f:
-        json.dump(data, f, indent=2, ensure_ascii=False)
-
-
 def enable_channel_config(config_path: Path, channel_name: str, defaults: dict[str, Any]) -> None:
-    data = read_config_data(config_path)
-    channels = data.setdefault("channels", {})
-    existing = channels.get(channel_name, {})
-    if not isinstance(existing, dict):
-        existing = {}
-    merged = merge_missing_defaults(existing, defaults)
-    merged["enabled"] = True
-    channels[channel_name] = merged
-    write_config_data(config_path, data)
+    def mutate(config: Config) -> None:
+        existing = getattr(config.channels, channel_name, {})
+        if not isinstance(existing, dict):
+            existing = {}
+        merged = merge_missing_defaults(existing, defaults)
+        merged["enabled"] = True
+        setattr(config.channels, channel_name, merged)
+
+    get_config_repository(config_path).update(mutate)
 
 
 def enable_feishu_instance_config(
@@ -297,24 +284,29 @@ def enable_feishu_instance_config(
     *,
     instance_id: str = DEFAULT_INSTANCE_ID,
 ) -> None:
-    data = read_config_data(config_path)
-    channels = data.setdefault("channels", {})
-    existing = channels.get("feishu", {})
-    if not isinstance(existing, dict):
-        existing = {}
-    channels["feishu"] = set_feishu_instance_enabled(existing, defaults, instance_id, True)
-    write_config_data(config_path, data)
+    def mutate(config: Config) -> None:
+        existing = getattr(config.channels, "feishu", {})
+        if not isinstance(existing, dict):
+            existing = {}
+        config.channels.feishu = set_feishu_instance_enabled(
+            existing,
+            defaults,
+            instance_id,
+            True,
+        )
+
+    get_config_repository(config_path).update(mutate)
 
 
 def disable_channel_config(config_path: Path, channel_name: str) -> None:
-    data = read_config_data(config_path)
-    channels = data.setdefault("channels", {})
-    existing = channels.get(channel_name, {})
-    if not isinstance(existing, dict):
-        existing = {}
-    existing["enabled"] = False
-    channels[channel_name] = existing
-    write_config_data(config_path, data)
+    def mutate(config: Config) -> None:
+        existing = getattr(config.channels, channel_name, {})
+        if not isinstance(existing, dict):
+            existing = {}
+        existing["enabled"] = False
+        setattr(config.channels, channel_name, existing)
+
+    get_config_repository(config_path).update(mutate)
 
 
 def disable_feishu_instance_config(
@@ -323,13 +315,18 @@ def disable_feishu_instance_config(
     *,
     instance_id: str = DEFAULT_INSTANCE_ID,
 ) -> None:
-    data = read_config_data(config_path)
-    channels = data.setdefault("channels", {})
-    existing = channels.get("feishu", {})
-    if not isinstance(existing, dict):
-        existing = {}
-    channels["feishu"] = set_feishu_instance_enabled(existing, defaults, instance_id, False)
-    write_config_data(config_path, data)
+    def mutate(config: Config) -> None:
+        existing = getattr(config.channels, "feishu", {})
+        if not isinstance(existing, dict):
+            existing = {}
+        config.channels.feishu = set_feishu_instance_enabled(
+            existing,
+            defaults,
+            instance_id,
+            False,
+        )
+
+    get_config_repository(config_path).update(mutate)
 
 
 def channel_enabled(config: Config, name: str) -> bool:
@@ -443,16 +440,16 @@ def optional_features_payload(
     last_action: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     from nanobot.channels.registry import discover_channel_names, discover_plugins
-    from nanobot.config.loader import load_config
+    from nanobot.config.loader import load_raw_config
 
     config_provided = config is not None
-    config = config or load_config()
+    config = config or load_raw_config()
     if not config_provided:
         with suppress(Exception):
             from nanobot.channels.feishu import refresh_saved_feishu_identities
 
             if refresh_saved_feishu_identities(config):
-                config = load_config()
+                config = load_raw_config()
     extras = optional_dependency_groups()
     builtin_channels = set(discover_channel_names())
     plugin_channels = discover_plugins()

nanobot/providers/factory.py

@@ -278,9 +278,9 @@ def load_provider_snapshot(
     *,
     preset_name: str | None = None,
 ) -> ProviderSnapshot:
-    from nanobot.config.loader import load_config, resolve_config_env_vars
+    from nanobot.config.loader import load_effective_config
 
     return build_provider_snapshot(
-        resolve_config_env_vars(load_config(config_path)),
+        load_effective_config(config_path),
         preset_name=preset_name,
     )

nanobot/webui/channel_connect.py

@@ -13,7 +13,7 @@ import httpx
 
 from nanobot.channels import feishu
 from nanobot.channels._feishu_instances import DEFAULT_INSTANCE_ID, validate_instance_id
-from nanobot.config.loader import load_config
+from nanobot.config.loader import load_raw_config
 
 
 class ChannelConnectError(Exception):
@@ -394,7 +394,7 @@ class WeixinConnectStore:
         from nanobot.bus.queue import MessageBus
         from nanobot.channels.weixin import WeixinChannel
 
-        section = getattr(load_config().channels, "weixin", None)
+        section = getattr(load_raw_config().channels, "weixin", None)
         if hasattr(section, "model_dump"):
             config = section.model_dump(mode="json", by_alias=True)
         elif isinstance(section, dict):

nanobot/webui/channel_validation.py

@@ -16,7 +16,7 @@ from typing import Any
 import httpx
 
 from nanobot.channels._setup import channel_setup_spec
-from nanobot.config.loader import load_config
+from nanobot.config.loader import load_raw_config
 from nanobot.security.network import resolve_url_target
 
 CheckStatus = str
@@ -42,7 +42,7 @@ def validate_channel_config(
     if not channel:
         return _payload("unknown", "unsupported", [_check("channel", "Channel", "fail", "Missing channel name")])
 
-    config = load_config()
+    config = load_raw_config()
     section = getattr(config.channels, channel, None)
     values = _channel_config(channel, section, instance_id=instance_id)
     values = _merge_form_values(channel, values, raw_values or {})

nanobot/webui/cli_apps_api.py

@@ -8,7 +8,7 @@ import time
 from typing import Any
 
 from nanobot.apps.cli import CliAppError, CliAppManager, CliAppsRuntimeConfig
-from nanobot.config.loader import load_config
+from nanobot.config.loader import load_raw_config
 
 QueryParams = dict[str, list[str]]
 
@@ -85,7 +85,7 @@ def _query_first(query: QueryParams, key: str) -> str | None:
 
 
 def _manager() -> CliAppManager:
-    config = load_config()
+    config = load_raw_config()
     cli_cfg = config.tools.cli_apps
     return CliAppManager(
         workspace=config.workspace_path,

nanobot/webui/mcp_presets_api.py

@@ -18,9 +18,9 @@ from typing import Any, Literal, Mapping
 
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.apps.protocol import app_manifest, compact_dict
-from nanobot.config.loader import load_config, resolve_config_env_vars, save_config
+from nanobot.config.loader import load_effective_config, load_raw_config, update_config
 from nanobot.config.paths import get_runtime_subdir
-from nanobot.config.schema import MCPServerConfig
+from nanobot.config.schema import Config, MCPServerConfig
 from nanobot.utils.helpers import ensure_dir
 
 QueryParams = dict[str, list[str]]
@@ -416,7 +416,7 @@ def _known_preset_names() -> set[str]:
 def _known_mcp_names() -> set[str]:
     names = _known_preset_names()
     with suppress(Exception):
-        names.update(load_config().tools.mcp_servers)
+        names.update(load_raw_config().tools.mcp_servers)
     return names
 
 
@@ -822,7 +822,7 @@ def mcp_presets_payload(
     last_action: dict[str, Any] | None = None,
     tool_preview: Mapping[str, list[str]] | None = None,
 ) -> dict[str, Any]:
-    config = load_config()
+    config = load_raw_config()
     known = _known_preset_names()
     preset_rows = [
         _preset_payload(preset, config.tools.mcp_servers)
@@ -921,7 +921,7 @@ async def mcp_presets_test_action(query: QueryParams) -> dict[str, Any]:
     display_name = _display_name_for(name, preset)
 
     try:
-        config = resolve_config_env_vars(load_config())
+        config = load_effective_config()
     except ValueError as exc:
         return mcp_presets_payload(last_action={
             "ok": False,
@@ -1180,19 +1180,16 @@ def _import_mcp_servers(raw_json: str | None) -> dict[str, MCPServerConfig]:
 
 
 def custom_mcp_action(action: str, query: QueryParams) -> dict[str, Any]:
-    config = load_config()
     if action == "custom":
         name, cfg = _custom_server_from_query(query)
-        config.tools.mcp_servers[name] = cfg
-        save_config(config)
+        update_config(lambda config: config.tools.mcp_servers.__setitem__(name, cfg))
         payload = mcp_presets_payload(last_action=_server_action_message(action, name))
         payload["requires_restart"] = True
         return payload
 
     if action in {"import", "import-cursor"}:
         servers = _import_mcp_servers(_query_first(query, "config"))
-        config.tools.mcp_servers.update(servers)
-        save_config(config)
+        update_config(lambda config: config.tools.mcp_servers.update(servers))
         payload = mcp_presets_payload(last_action={
             "ok": True,
             "message": f"Imported {len(servers)} MCP server(s).",
@@ -1202,12 +1199,15 @@ def custom_mcp_action(action: str, query: QueryParams) -> dict[str, Any]:
 
     if action == "tools":
         name = _validated_server_name((_query_first(query, "name") or "").strip())
-        cfg = config.tools.mcp_servers.get(name)
-        if cfg is None:
-            raise McpPresetError("unknown MCP server", status=404)
-        cfg.enabled_tools = _parse_enabled_tools(_query_first(query, "enabled_tools"))
-        config.tools.mcp_servers[name] = cfg
-        save_config(config)
+
+        def mutate(config: Config) -> None:
+            cfg = config.tools.mcp_servers.get(name)
+            if cfg is None:
+                raise McpPresetError("unknown MCP server", status=404)
+            cfg.enabled_tools = _parse_enabled_tools(_query_first(query, "enabled_tools"))
+            config.tools.mcp_servers[name] = cfg
+
+        update_config(mutate)
         payload = mcp_presets_payload(last_action=_server_action_message(action, name))
         payload["requires_restart"] = True
         return payload
@@ -1221,31 +1221,42 @@ def mcp_presets_action(action: str, query: QueryParams) -> dict[str, Any]:
         raise McpPresetError("missing MCP preset name")
     preset = _preset_by_name_optional(name)
 
-    config = load_config()
-    existing = config.tools.mcp_servers.get(name)
-
     if action == "enable":
         if preset is None:
             raise McpPresetError("unknown MCP preset", status=404)
-        config.tools.mcp_servers[preset.name] = _materialize_server(preset, query, existing)
-        save_config(config)
+
+        def mutate(config: Config) -> None:
+            existing = config.tools.mcp_servers.get(name)
+            config.tools.mcp_servers[preset.name] = _materialize_server(
+                preset,
+                query,
+                existing,
+            )
+
+        update_config(mutate)
         payload = mcp_presets_payload(last_action=_action_message(action, preset))
         payload["requires_restart"] = True
         return payload
 
     if action == "remove":
-        if preset is None and name not in config.tools.mcp_servers:
-            raise McpPresetError("unknown MCP server", status=404)
         removed_runtime_files = False
         cleanup_error = ""
-        if name in config.tools.mcp_servers:
-            existing_cfg = config.tools.mcp_servers[name]
+        removed: dict[str, MCPServerConfig] = {}
+
+        def mutate(config: Config) -> None:
+            existing_cfg = config.tools.mcp_servers.get(name)
+            if preset is None and existing_cfg is None:
+                raise McpPresetError("unknown MCP server", status=404)
+            if existing_cfg is not None:
+                removed[name] = existing_cfg
+                del config.tools.mcp_servers[name]
+
+        update_config(mutate)
+        if existing_cfg := removed.get(name):
             try:
                 removed_runtime_files = _remove_managed_stdio_cwd(name, existing_cfg)
             except OSError as exc:
                 cleanup_error = str(exc)
-            del config.tools.mcp_servers[name]
-            save_config(config)
         last_action = (
             _action_message(action, preset)
             if preset is not None

nanobot/webui/settings_api.py

@@ -22,8 +22,13 @@ from nanobot.audio.transcription_registry import (
     resolve_transcription_provider,
     transcription_provider_names,
 )
-from nanobot.config.loader import get_config_path, load_config, resolve_config_env_vars, save_config
-from nanobot.config.schema import ModelPresetConfig, ProviderConfig
+from nanobot.config.loader import (
+    get_config_path,
+    load_effective_config,
+    load_raw_config,
+    update_config,
+)
+from nanobot.config.schema import Config, ModelPresetConfig, ProviderConfig
 from nanobot.providers.image_generation import (
     get_image_gen_provider,
     image_gen_provider_names,
@@ -512,7 +517,7 @@ def provider_models_payload(query: QueryParams) -> dict[str, Any]:
     if not provider_name:
         raise WebUISettingsError("provider is required")
 
-    config = load_config()
+    config = load_raw_config()
     resolved_provider = _resolve_settings_provider(config, provider_name)
     if resolved_provider is None:
         raise WebUISettingsError("unknown provider")
@@ -749,7 +754,7 @@ def settings_payload(
     restart_required_sections: list[str] | None = None,
     apply_state: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
-    config = load_config()
+    config = load_raw_config()
     defaults = config.agents.defaults
     active_preset_name = defaults.model_preset or "default"
     try:
@@ -956,105 +961,91 @@ def settings_payload(
 
 def settings_usage_payload() -> dict[str, Any]:
     """Return the lightweight token usage slice for Overview refreshes."""
-    config = load_config()
+    config = load_raw_config()
     return token_usage_payload(timezone_name=config.agents.defaults.timezone)
 
 
 def update_agent_settings(query: QueryParams) -> dict[str, Any]:
-    config = load_config()
-    defaults = config.agents.defaults
-    changed = False
     restart_required = False
 
-    if "model_preset" in query or "modelPreset" in query:
-        preset = (_query_first_alias(query, "model_preset", "modelPreset") or "").strip()
-        preset_value = None if not preset or preset == "default" else preset
-        if preset_value is not None and preset_value not in config.model_presets:
-            raise WebUISettingsError("unknown model preset")
-        if defaults.model_preset != preset_value:
+    def mutate(config: Config) -> None:
+        nonlocal restart_required
+        defaults = config.agents.defaults
+
+        if "model_preset" in query or "modelPreset" in query:
+            preset = (_query_first_alias(query, "model_preset", "modelPreset") or "").strip()
+            preset_value = None if not preset or preset == "default" else preset
+            if preset_value is not None and preset_value not in config.model_presets:
+                raise WebUISettingsError("unknown model preset")
             defaults.model_preset = preset_value
-            changed = True
 
-    model = _query_first(query, "model")
-    if model is not None:
-        model = model.strip()
-        if not model:
-            raise WebUISettingsError("model is required")
-        if defaults.model != model:
+        model = _query_first(query, "model")
+        if model is not None:
+            model = model.strip()
+            if not model:
+                raise WebUISettingsError("model is required")
             defaults.model = model
-            changed = True
 
-    provider = _query_first(query, "provider")
-    if provider is not None:
-        provider = provider.strip()
-        if not provider:
-            raise WebUISettingsError("provider is required")
-        _validate_configured_provider(config, provider)
-        if defaults.provider != provider:
+        provider = _query_first(query, "provider")
+        if provider is not None:
+            provider = provider.strip()
+            if not provider:
+                raise WebUISettingsError("provider is required")
+            _validate_configured_provider(config, provider)
             defaults.provider = provider
-            changed = True
 
-    context_window_tokens = _parse_context_window_tokens(
-        _query_first_alias(query, "context_window_tokens", "contextWindowTokens")
-    )
-    if (
-        context_window_tokens is not None
-        and defaults.context_window_tokens != context_window_tokens
-    ):
-        defaults.context_window_tokens = context_window_tokens
-        changed = True
+        context_window_tokens = _parse_context_window_tokens(
+            _query_first_alias(query, "context_window_tokens", "contextWindowTokens")
+        )
+        if context_window_tokens is not None:
+            defaults.context_window_tokens = context_window_tokens
 
-    timezone = _query_first(query, "timezone")
-    if timezone is not None:
-        timezone = timezone.strip()
-        if not timezone:
-            raise WebUISettingsError("timezone is required")
-        try:
-            ZoneInfo(timezone)
-        except Exception:
-            raise WebUISettingsError("invalid timezone") from None
-        if defaults.timezone != timezone:
-            defaults.timezone = timezone
-            changed = True
-            restart_required = True
+        timezone = _query_first(query, "timezone")
+        if timezone is not None:
+            timezone = timezone.strip()
+            if not timezone:
+                raise WebUISettingsError("timezone is required")
+            try:
+                ZoneInfo(timezone)
+            except Exception:
+                raise WebUISettingsError("invalid timezone") from None
+            if defaults.timezone != timezone:
+                defaults.timezone = timezone
+                restart_required = True
 
-    bot_name = _query_first_alias(query, "bot_name", "botName")
-    if bot_name is not None:
-        bot_name = bot_name.strip()
-        if not bot_name:
-            raise WebUISettingsError("bot_name is required")
-        if defaults.bot_name != bot_name:
-            defaults.bot_name = bot_name
-            changed = True
-            restart_required = True
+        bot_name = _query_first_alias(query, "bot_name", "botName")
+        if bot_name is not None:
+            bot_name = bot_name.strip()
+            if not bot_name:
+                raise WebUISettingsError("bot_name is required")
+            if defaults.bot_name != bot_name:
+                defaults.bot_name = bot_name
+                restart_required = True
 
-    bot_icon = _query_first_alias(query, "bot_icon", "botIcon")
-    if bot_icon is not None:
-        bot_icon = bot_icon.strip()
-        if defaults.bot_icon != bot_icon:
-            defaults.bot_icon = bot_icon
-            changed = True
-            restart_required = True
+        bot_icon = _query_first_alias(query, "bot_icon", "botIcon")
+        if bot_icon is not None:
+            bot_icon = bot_icon.strip()
+            if defaults.bot_icon != bot_icon:
+                defaults.bot_icon = bot_icon
+                restart_required = True
 
-    tool_hint_max_length = _query_first_alias(
-        query,
-        "tool_hint_max_length",
-        "toolHintMaxLength",
-    )
-    if tool_hint_max_length is not None:
-        try:
-            parsed = int(tool_hint_max_length)
-        except ValueError:
-            raise WebUISettingsError("tool_hint_max_length must be an integer") from None
-        if parsed < 20 or parsed > 500:
-            raise WebUISettingsError("tool_hint_max_length must be between 20 and 500")
-        if defaults.tool_hint_max_length != parsed:
-            defaults.tool_hint_max_length = parsed
-            changed = True
-            restart_required = True
+        tool_hint_max_length = _query_first_alias(
+            query,
+            "tool_hint_max_length",
+            "toolHintMaxLength",
+        )
+        if tool_hint_max_length is not None:
+            try:
+                parsed = int(tool_hint_max_length)
+            except ValueError:
+                raise WebUISettingsError("tool_hint_max_length must be an integer") from None
+            if parsed < 20 or parsed > 500:
+                raise WebUISettingsError("tool_hint_max_length must be between 20 and 500")
+            if defaults.tool_hint_max_length != parsed:
+                defaults.tool_hint_max_length = parsed
+                restart_required = True
 
-    if changed:
-        save_config(config)
+    update_config(mutate)
     return settings_payload(requires_restart=restart_required)
 
 
@@ -1072,23 +1063,23 @@ def create_model_configuration(query: QueryParams) -> dict[str, Any]:
         raise WebUISettingsError("provider is required")
 
     name = _model_configuration_slug(raw_name or label)
-    config = load_config()
-    if name in config.model_presets:
-        raise WebUISettingsError("configuration already exists", status=409)
-    _validate_configured_provider(config, provider)
+    def mutate(config: Config) -> None:
+        if name in config.model_presets:
+            raise WebUISettingsError("configuration already exists", status=409)
+        _validate_configured_provider(config, provider)
+        base = config.resolve_default_preset()
+        config.model_presets[name] = ModelPresetConfig(
+            label=label,
+            model=model,
+            provider=provider,
+            max_tokens=base.max_tokens,
+            context_window_tokens=base.context_window_tokens,
+            temperature=base.temperature,
+            reasoning_effort=base.reasoning_effort,
+        )
+        config.agents.defaults.model_preset = name
 
-    base = config.resolve_default_preset()
-    config.model_presets[name] = ModelPresetConfig(
-        label=label,
-        model=model,
-        provider=provider,
-        max_tokens=base.max_tokens,
-        context_window_tokens=base.context_window_tokens,
-        temperature=base.temperature,
-        reasoning_effort=base.reasoning_effort,
-    )
-    config.agents.defaults.model_preset = name
-    save_config(config)
+    update_config(mutate)
     return settings_payload()
 
 
@@ -1097,56 +1088,41 @@ def update_model_configuration(query: QueryParams) -> dict[str, Any]:
     if not name or name == "default":
         raise WebUISettingsError("model configuration is required")
 
-    config = load_config()
-    preset = config.model_presets.get(name)
-    if preset is None:
-        raise WebUISettingsError("unknown model configuration")
+    def mutate(config: Config) -> None:
+        preset = config.model_presets.get(name)
+        if preset is None:
+            raise WebUISettingsError("unknown model configuration")
 
-    changed = False
-    label = _query_first_alias(query, "label", "displayName")
-    if label is not None:
-        label = label.strip()
-        if not label:
-            raise WebUISettingsError("label is required")
-        if preset.label != label:
+        label = _query_first_alias(query, "label", "displayName")
+        if label is not None:
+            label = label.strip()
+            if not label:
+                raise WebUISettingsError("label is required")
             preset.label = label
-            changed = True
 
-    model = _query_first(query, "model")
-    if model is not None:
-        model = model.strip()
-        if not model:
-            raise WebUISettingsError("model is required")
-        if preset.model != model:
+        model = _query_first(query, "model")
+        if model is not None:
+            model = model.strip()
+            if not model:
+                raise WebUISettingsError("model is required")
             preset.model = model
-            changed = True
 
-    provider = _query_first(query, "provider")
-    if provider is not None:
-        provider = provider.strip()
-        if not provider:
-            raise WebUISettingsError("provider is required")
-        _validate_configured_provider(config, provider)
-        if preset.provider != provider:
+        provider = _query_first(query, "provider")
+        if provider is not None:
+            provider = provider.strip()
+            if not provider:
+                raise WebUISettingsError("provider is required")
+            _validate_configured_provider(config, provider)
             preset.provider = provider
-            changed = True
 
-    context_window_tokens = _parse_context_window_tokens(
-        _query_first_alias(query, "context_window_tokens", "contextWindowTokens")
-    )
-    if (
-        context_window_tokens is not None
-        and preset.context_window_tokens != context_window_tokens
-    ):
-        preset.context_window_tokens = context_window_tokens
-        changed = True
-
-    if config.agents.defaults.model_preset != name:
+        context_window_tokens = _parse_context_window_tokens(
+            _query_first_alias(query, "context_window_tokens", "contextWindowTokens")
+        )
+        if context_window_tokens is not None:
+            preset.context_window_tokens = context_window_tokens
         config.agents.defaults.model_preset = name
-        changed = True
 
-    if changed:
-        save_config(config)
+    update_config(mutate)
     return settings_payload()
 
 
@@ -1155,43 +1131,37 @@ def update_provider_settings(query: QueryParams) -> dict[str, Any]:
     if not provider_name:
         raise WebUISettingsError("provider is required")
 
-    config = load_config()
-    resolved_provider = _resolve_settings_provider(config, provider_name)
-    if resolved_provider is None:
-        raise WebUISettingsError("unknown provider")
-    spec, provider_key, provider_config = resolved_provider
-    if spec.is_oauth:
-        raise WebUISettingsError("unknown provider")
+    provider_key = ""
 
-    changed = False
-    if "api_key" in query or "apiKey" in query:
-        api_key = _query_first_alias(query, "api_key", "apiKey")
-        api_key = (api_key or "").strip() or None
-        if provider_config.api_key != api_key:
-            provider_config.api_key = api_key
-            changed = True
+    def mutate(config: Config) -> None:
+        nonlocal provider_key
+        resolved_provider = _resolve_settings_provider(config, provider_name)
+        if resolved_provider is None:
+            raise WebUISettingsError("unknown provider")
+        spec, provider_key, provider_config = resolved_provider
+        if spec.is_oauth:
+            raise WebUISettingsError("unknown provider")
 
-    if "api_base" in query or "apiBase" in query:
-        api_base = _query_first_alias(query, "api_base", "apiBase")
-        api_base = (api_base or "").strip() or None
-        if provider_config.api_base != api_base:
-            provider_config.api_base = api_base
-            changed = True
+        if "api_key" in query or "apiKey" in query:
+            api_key = _query_first_alias(query, "api_key", "apiKey")
+            provider_config.api_key = (api_key or "").strip() or None
 
-    if "api_type" in query:
-        if spec.name == "openai":
+        if "api_base" in query or "apiBase" in query:
+            api_base = _query_first_alias(query, "api_base", "apiBase")
+            provider_config.api_base = (api_base or "").strip() or None
+
+        if "api_type" in query and spec.name == "openai":
             api_type = (_query_first(query, "api_type") or "").strip()
             try:
-                parsed_api_type = type(provider_config)(api_type=api_type).api_type
+                provider_config.api_type = type(provider_config)(api_type=api_type).api_type
             except Exception:
-                raise WebUISettingsError("api_type must be auto, chat_completions, or responses") from None
-            if provider_config.api_type != parsed_api_type:
-                provider_config.api_type = parsed_api_type
-                changed = True
+                raise WebUISettingsError(
+                    "api_type must be auto, chat_completions, or responses"
+                ) from None
 
-    if changed:
-        save_config(config)
-    image_config = config.tools.image_generation
+    commit = update_config(mutate)
+    changed = bool(commit.changed_paths)
+    image_config = commit.after.config.tools.image_generation
     restart_required = (
         changed
         and image_config.enabled
@@ -1218,7 +1188,7 @@ def login_oauth_provider(query: QueryParams) -> dict[str, Any]:
             ) from None
 
         try:
-            proxy = resolve_config_env_vars(load_config()).providers.openai_codex.proxy or None
+            proxy = load_effective_config().providers.openai_codex.proxy or None
         except ValueError as e:
             raise WebUISettingsError(str(e), status=400) from e
         token = None
@@ -1299,16 +1269,18 @@ def update_network_safety_settings(query: QueryParams) -> dict[str, Any]:
     if raw_allow is None and raw_default_access_mode is None:
         raise WebUISettingsError("webui_allow_local_service_access or webui_default_access_mode is required")
 
-    config = load_config()
-    changed = False
     if raw_allow is not None:
         webui_allow_local_service_access = _parse_bool(raw_allow, "webui_allow_local_service_access")
-        if config.tools.webui_allow_local_service_access != webui_allow_local_service_access:
-            config.tools.webui_allow_local_service_access = webui_allow_local_service_access
-            changed = True
-
-    if changed:
-        save_config(config)
+        commit = update_config(
+            lambda config: setattr(
+                config.tools,
+                "webui_allow_local_service_access",
+                webui_allow_local_service_access,
+            )
+        )
+        changed = bool(commit.changed_paths)
+    else:
+        changed = False
     if raw_default_access_mode is not None:
         default_access_mode = raw_default_access_mode.strip().lower()
         if default_access_mode == "restricted":
@@ -1328,134 +1300,120 @@ def update_web_search_settings(query: QueryParams) -> dict[str, Any]:
     if provider_option is None:
         raise WebUISettingsError("unknown web search provider")
 
-    config = load_config()
-    search_config = config.tools.web.search
-    web_config = config.tools.web
-    previous_provider = search_config.provider
-    changed = False
     restart_required = False
 
-    def set_search_value(attr: str, value: object) -> None:
-        nonlocal changed
-        if getattr(search_config, attr) != value:
-            setattr(search_config, attr, value)
-            changed = True
-
-    def set_fetch_value(attr: str, value: object) -> None:
-        nonlocal changed
-        if getattr(web_config.fetch, attr) != value:
-            setattr(web_config.fetch, attr, value)
-            changed = True
-
-    if search_config.provider != provider_name:
+    def mutate(config: Config) -> None:
+        nonlocal restart_required
+        search_config = config.tools.web.search
+        web_config = config.tools.web
+        previous_provider = search_config.provider
         search_config.provider = provider_name
-        changed = True
 
-    credential = provider_option["credential"]
-    if credential == "none":
-        set_search_value("api_key", "")
-        set_search_value("base_url", "")
-    elif credential == "base_url":
-        base_url = _query_first_alias(query, "base_url", "baseUrl")
-        base_url = base_url.strip() if base_url is not None else None
-        if not base_url and previous_provider == provider_name and search_config.base_url:
-            base_url = search_config.base_url
-        if not base_url:
-            raise WebUISettingsError("base_url is required")
-        set_search_value("base_url", base_url)
-        set_search_value("api_key", "")
-    elif credential in {"api_key", "optional_api_key"}:
-        raw_api_key = _query_first_alias(query, "api_key", "apiKey")
-        api_key = raw_api_key.strip() if raw_api_key is not None else None
-        if api_key is None and previous_provider == provider_name and search_config.api_key:
-            api_key = search_config.api_key
-        if credential == "api_key" and not api_key:
-            raise WebUISettingsError("api_key is required")
-        set_search_value("api_key", api_key or "")
-        set_search_value("base_url", "")
-    else:
-        raise WebUISettingsError("unknown web search credential type")
+        credential = provider_option["credential"]
+        if credential == "none":
+            search_config.api_key = ""
+            search_config.base_url = ""
+        elif credential == "base_url":
+            base_url = _query_first_alias(query, "base_url", "baseUrl")
+            base_url = base_url.strip() if base_url is not None else None
+            if not base_url and previous_provider == provider_name and search_config.base_url:
+                base_url = search_config.base_url
+            if not base_url:
+                raise WebUISettingsError("base_url is required")
+            search_config.base_url = base_url
+            search_config.api_key = ""
+        elif credential in {"api_key", "optional_api_key"}:
+            raw_api_key = _query_first_alias(query, "api_key", "apiKey")
+            api_key = raw_api_key.strip() if raw_api_key is not None else None
+            if api_key is None and previous_provider == provider_name and search_config.api_key:
+                api_key = search_config.api_key
+            if credential == "api_key" and not api_key:
+                raise WebUISettingsError("api_key is required")
+            search_config.api_key = api_key or ""
+            search_config.base_url = ""
+        else:
+            raise WebUISettingsError("unknown web search credential type")
 
-    max_results = _query_first_alias(query, "max_results", "maxResults")
-    if max_results is not None:
-        try:
-            parsed = int(max_results)
-        except ValueError:
-            raise WebUISettingsError("max_results must be an integer") from None
-        if parsed < 1 or parsed > 10:
-            raise WebUISettingsError("max_results must be between 1 and 10")
-        set_search_value("max_results", parsed)
+        max_results = _query_first_alias(query, "max_results", "maxResults")
+        if max_results is not None:
+            try:
+                parsed = int(max_results)
+            except ValueError:
+                raise WebUISettingsError("max_results must be an integer") from None
+            if parsed < 1 or parsed > 10:
+                raise WebUISettingsError("max_results must be between 1 and 10")
+            search_config.max_results = parsed
 
-    timeout = _query_first(query, "timeout")
-    if timeout is not None:
-        try:
-            parsed_timeout = int(timeout)
-        except ValueError:
-            raise WebUISettingsError("timeout must be an integer") from None
-        if parsed_timeout < 1 or parsed_timeout > 120:
-            raise WebUISettingsError("timeout must be between 1 and 120")
-        set_search_value("timeout", parsed_timeout)
+        timeout = _query_first(query, "timeout")
+        if timeout is not None:
+            try:
+                parsed_timeout = int(timeout)
+            except ValueError:
+                raise WebUISettingsError("timeout must be an integer") from None
+            if parsed_timeout < 1 or parsed_timeout > 120:
+                raise WebUISettingsError("timeout must be between 1 and 120")
+            search_config.timeout = parsed_timeout
 
-    use_jina_reader = _query_first_alias(query, "use_jina_reader", "useJinaReader")
-    if use_jina_reader is not None:
-        normalized = use_jina_reader.strip().lower()
-        if normalized not in {"1", "0", "true", "false", "yes", "no"}:
-            raise WebUISettingsError("use_jina_reader must be boolean")
-        previous_jina_reader = web_config.fetch.use_jina_reader
-        set_fetch_value("use_jina_reader", normalized in {"1", "true", "yes"})
-        if web_config.fetch.use_jina_reader != previous_jina_reader:
-            restart_required = True
+        use_jina_reader = _query_first_alias(query, "use_jina_reader", "useJinaReader")
+        if use_jina_reader is not None:
+            normalized = use_jina_reader.strip().lower()
+            if normalized not in {"1", "0", "true", "false", "yes", "no"}:
+                raise WebUISettingsError("use_jina_reader must be boolean")
+            enabled = normalized in {"1", "true", "yes"}
+            if web_config.fetch.use_jina_reader != enabled:
+                web_config.fetch.use_jina_reader = enabled
+                restart_required = True
 
-    if changed:
-        save_config(config)
+    update_config(mutate)
     return settings_payload(requires_restart=restart_required)
 
 
 def update_api_settings(query: QueryParams) -> dict[str, Any]:
     """Update the managed OpenAI-compatible API configuration."""
-    config = load_config()
-    api = config.api
+    def mutate(config: Config) -> None:
+        api = config.api
 
-    host = _query_first(query, "host")
-    if host is not None:
-        host = host.strip()
-        if not host:
-            raise WebUISettingsError("host is required")
-        api.host = host
+        host = _query_first(query, "host")
+        if host is not None:
+            host = host.strip()
+            if not host:
+                raise WebUISettingsError("host is required")
+            api.host = host
 
-    port = _query_first(query, "port")
-    if port is not None:
-        try:
-            parsed_port = int(port)
-        except ValueError:
-            raise WebUISettingsError("port must be an integer") from None
-        if parsed_port < 1 or parsed_port > 65535:
-            raise WebUISettingsError("port must be between 1 and 65535")
-        api.port = parsed_port
+        port = _query_first(query, "port")
+        if port is not None:
+            try:
+                parsed_port = int(port)
+            except ValueError:
+                raise WebUISettingsError("port must be an integer") from None
+            if parsed_port < 1 or parsed_port > 65535:
+                raise WebUISettingsError("port must be between 1 and 65535")
+            api.port = parsed_port
 
-    timeout = _query_first(query, "timeout")
-    if timeout is not None:
-        try:
-            parsed_timeout = float(timeout)
-        except ValueError:
-            raise WebUISettingsError("timeout must be a number") from None
-        if parsed_timeout < 1 or parsed_timeout > 3600:
-            raise WebUISettingsError("timeout must be between 1 and 3600")
-        api.timeout = parsed_timeout
+        timeout = _query_first(query, "timeout")
+        if timeout is not None:
+            try:
+                parsed_timeout = float(timeout)
+            except ValueError:
+                raise WebUISettingsError("timeout must be a number") from None
+            if parsed_timeout < 1 or parsed_timeout > 3600:
+                raise WebUISettingsError("timeout must be between 1 and 3600")
+            api.timeout = parsed_timeout
 
-    api_key = _query_first_alias(query, "api_key", "apiKey")
-    if api_key is not None:
-        api.api_key = api_key.strip()
+        api_key = _query_first_alias(query, "api_key", "apiKey")
+        if api_key is not None:
+            api.api_key = api_key.strip()
 
-    if not is_loopback_host(api.host) and not api.api_key.strip():
-        raise WebUISettingsError("an API key is required when the API is available on the network")
+        if not is_loopback_host(api.host) and not api.api_key.strip():
+            raise WebUISettingsError(
+                "an API key is required when the API is available on the network"
+            )
 
-    save_config(config)
+    update_config(mutate)
     return settings_payload()
 
 
-def update_image_generation_settings(query: QueryParams) -> dict[str, Any]:
-    config = load_config()
+def _apply_image_generation_settings(config: Config, query: QueryParams) -> bool:
     image_config = config.tools.image_generation
     changed = False
 
@@ -1547,13 +1505,21 @@ def update_image_generation_settings(query: QueryParams) -> dict[str, Any]:
         if not selected_provider or not selected_provider["configured"]:
             raise WebUISettingsError("image generation provider is not configured")
 
-    if changed:
-        save_config(config)
+    return changed
+
+
+def update_image_generation_settings(query: QueryParams) -> dict[str, Any]:
+    changed = False
+
+    def mutate(config: Config) -> None:
+        nonlocal changed
+        changed = _apply_image_generation_settings(config, query)
+
+    update_config(mutate)
     return settings_payload(requires_restart=changed)
 
 
-def update_transcription_settings(query: QueryParams) -> dict[str, Any]:
-    config = load_config()
+def _apply_transcription_settings(config: Config, query: QueryParams) -> bool:
     transcription = config.transcription
     changed = False
 
@@ -1617,6 +1583,12 @@ def update_transcription_settings(query: QueryParams) -> dict[str, Any]:
             transcription.max_upload_mb = parsed_upload
             changed = True
 
-    if changed:
-        save_config(config)
+    return changed
+
+
+def update_transcription_settings(query: QueryParams) -> dict[str, Any]:
+    def mutate(config: Config) -> None:
+        _apply_transcription_settings(config, query)
+
+    update_config(mutate)
     return settings_payload()

nanobot/webui/settings_routes.py

@@ -21,7 +21,7 @@ from nanobot.agent.tools.mcp import request_mcp_reload
 from nanobot.api.runtime import ApiRuntime, ApiStartOptions, api_runtime_paths
 from nanobot.bus.queue import MessageBus
 from nanobot.channels._setup import channel_setup_spec
-from nanobot.config.loader import get_config_path, load_config, save_config
+from nanobot.config.loader import get_config_path, load_raw_config, update_config
 from nanobot.optional_features import (
     OptionalFeatureError,
     extra_installed,
@@ -397,7 +397,7 @@ class WebUISettingsRouter:
                 allow_install=self._allow_feature_package_install(connection, request),
             )
             update_api_settings(self._parse_api_service_settings_query(request))
-            config = load_config()
+            config = load_raw_config()
             runtime = self._api_runtime()
             options = ApiStartOptions(
                 host=config.api.host,
@@ -465,7 +465,7 @@ class WebUISettingsRouter:
         return ApiRuntime(paths=api_runtime_paths(config_path))
 
     def _api_service_payload(self, *, last_action: str | None = None) -> dict[str, Any]:
-        config = load_config()
+        config = load_raw_config()
         status = self._api_runtime().status()
         extras = optional_dependency_groups()
         connect_host = "127.0.0.1" if config.api.host in {"0.0.0.0", "::"} else config.api.host
@@ -773,51 +773,56 @@ class WebUISettingsRouter:
         if not raw_values:
             return []
 
-        config = load_config()
-        section = getattr(config.channels, name, None)
-        if name == "feishu":
-            from nanobot.channels._feishu_instances import feishu_instance_specs
-            from nanobot.channels.feishu import FeishuChannel
-
-            specs = feishu_instance_specs(section, FeishuChannel.default_config())
-            selected = next((spec for spec in specs if spec.instance_id == instance_id), None)
-            channel_config = dict(selected.config) if selected is not None else {}
-        elif hasattr(section, "model_dump"):
-            channel_config = section.model_dump(mode="json", by_alias=True)
-        elif isinstance(section, dict):
-            channel_config = dict(section)
-        else:
-            channel_config = {}
-
         saved: list[str] = []
-        prefix = f"channels.{name}."
-        for raw_key, raw_value in raw_values.items():
-            if not isinstance(raw_key, str) or not raw_key:
-                raise WebUISettingsError("channel settings payload contains an invalid key")
-            field = raw_key[len(prefix):] if raw_key.startswith(prefix) else raw_key
-            value_type = field_types.get(field)
-            if value_type is None:
-                raise WebUISettingsError(f"'{raw_key}' cannot be configured from WebUI")
-            value = self._coerce_channel_value(raw_key, raw_value, value_type)
-            if value is _SKIP_FIELD:
-                continue
-            self._assign_channel_config_value(channel_config, field, value)
-            saved.append(raw_key)
 
-        if name == "feishu":
-            from nanobot.channels._feishu_instances import upsert_feishu_instance
-            from nanobot.channels.feishu import FeishuChannel
+        def mutate(config: Any) -> None:
+            section = getattr(config.channels, name, None)
+            if name == "feishu":
+                from nanobot.channels._feishu_instances import feishu_instance_specs
+                from nanobot.channels.feishu import FeishuChannel
 
-            existing = getattr(config.channels, name, None)
-            channel_config = upsert_feishu_instance(
-                existing if isinstance(existing, dict) else {},
-                FeishuChannel.default_config(),
-                instance_id,
-                channel_config,
-            )
+                specs = feishu_instance_specs(section, FeishuChannel.default_config())
+                selected = next(
+                    (spec for spec in specs if spec.instance_id == instance_id),
+                    None,
+                )
+                channel_config = dict(selected.config) if selected is not None else {}
+            elif hasattr(section, "model_dump"):
+                channel_config = section.model_dump(mode="json", by_alias=True)
+            elif isinstance(section, dict):
+                channel_config = dict(section)
+            else:
+                channel_config = {}
 
-        setattr(config.channels, name, channel_config)
-        save_config(config)
+            prefix = f"channels.{name}."
+            for raw_key, raw_value in raw_values.items():
+                if not isinstance(raw_key, str) or not raw_key:
+                    raise WebUISettingsError("channel settings payload contains an invalid key")
+                field = raw_key[len(prefix):] if raw_key.startswith(prefix) else raw_key
+                value_type = field_types.get(field)
+                if value_type is None:
+                    raise WebUISettingsError(f"'{raw_key}' cannot be configured from WebUI")
+                value = self._coerce_channel_value(raw_key, raw_value, value_type)
+                if value is _SKIP_FIELD:
+                    continue
+                self._assign_channel_config_value(channel_config, field, value)
+                saved.append(raw_key)
+
+            if name == "feishu":
+                from nanobot.channels._feishu_instances import upsert_feishu_instance
+                from nanobot.channels.feishu import FeishuChannel
+
+                existing = getattr(config.channels, name, None)
+                channel_config = upsert_feishu_instance(
+                    existing if isinstance(existing, dict) else {},
+                    FeishuChannel.default_config(),
+                    instance_id,
+                    channel_config,
+                )
+
+            setattr(config.channels, name, channel_config)
+
+        update_config(mutate)
         return saved
 
     @staticmethod
@@ -1058,7 +1063,7 @@ class WebUISettingsRouter:
         if _is_local_browser_request(connection, request.headers):
             return True
         try:
-            return bool(load_config().tools.webui_allow_remote_package_install)
+            return bool(load_raw_config().tools.webui_allow_remote_package_install)
         except Exception:
             self.logger.exception("failed to load remote package install policy")
             return False

nanobot/webui/transcription_ws.py

@@ -13,7 +13,7 @@ from nanobot.audio.transcription import (
     resolve_transcription_config,
     transcribe_audio_data_url,
 )
-from nanobot.config.loader import load_config
+from nanobot.config.loader import load_raw_config
 
 _MAX_REQUEST_ID_LENGTH = 80
 
@@ -38,7 +38,7 @@ async def webui_transcription_event(envelope: dict[str, Any]) -> tuple[str, dict
     try:
         text = await transcribe_audio_data_url(
             envelope.get("data_url"),
-            resolve_transcription_config(load_config()),
+            resolve_transcription_config(load_raw_config()),
             duration_ms=envelope.get("duration_ms"),
         )
     except TranscriptionIngressError as exc:

nanobot/webui/ws_http.py

@@ -107,8 +107,8 @@ def _decode_api_key(raw_key: str) -> str | None:
 
 def _default_model_name_from_config() -> str | None:
     try:
-        from nanobot.config.loader import load_config
-        model = load_config().resolve_preset().model.strip()
+        from nanobot.config.loader import load_raw_config
+        model = load_raw_config().resolve_preset().model.strip()
         return model or None
     except Exception as e:
         logger.debug("bootstrap model_name could not load from config: {}", e)

tests/agent/test_mcp_connection.py

@@ -20,7 +20,7 @@ from nanobot.agent.tools import mcp as mcp_runtime
 from nanobot.agent.tools.base import Tool
 from nanobot.agent.tools.mcp import MCPResourceWrapper, MCPToolWrapper
 from nanobot.bus.queue import MessageBus
-from nanobot.config.loader import load_config, save_config
+from nanobot.config.loader import load_raw_config, save_config
 from nanobot.config.schema import MCPServerConfig
 
 
@@ -201,7 +201,7 @@ async def test_reload_mcp_servers_adds_and_removes_tools_without_restart(
 ):
     config_path = tmp_path / "config.json"
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
-    config = load_config()
+    config = load_raw_config()
     config.tools.mcp_servers["browserbase"] = MCPServerConfig(
         type="stdio",
         command="browserbase-mcp",
@@ -233,7 +233,7 @@ async def test_reload_mcp_servers_adds_and_removes_tools_without_restart(
     assert loop.tools.has("mcp_browserbase_navigate")
     assert "browserbase" in loop._mcp_stacks
 
-    config = load_config()
+    config = load_raw_config()
     del config.tools.mcp_servers["browserbase"]
     save_config(config)
 
@@ -253,7 +253,7 @@ async def test_request_mcp_reload_reaches_runtime_control_without_restart(
 ):
     config_path = tmp_path / "config.json"
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
-    config = load_config()
+    config = load_raw_config()
     config.tools.mcp_servers["browserbase"] = MCPServerConfig(
         type="stdio",
         command="browserbase-mcp",
@@ -292,7 +292,7 @@ async def test_request_mcp_reload_reaches_runtime_control_without_restart(
     assert result["requires_restart"] is False
     assert loop.tools.has("mcp_browserbase_navigate")
 
-    config = load_config()
+    config = load_raw_config()
     del config.tools.mcp_servers["browserbase"]
     save_config(config)
 
@@ -314,7 +314,7 @@ async def test_reload_mcp_servers_retries_configured_server_without_live_stack(
 ):
     config_path = tmp_path / "config.json"
     monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path)
-    config = load_config()
+    config = load_raw_config()
     config.tools.mcp_servers["browserbase"] = MCPServerConfig(
         type="stdio",
         command="browserbase-mcp",

tests/channels/test_channel_manager_hot_reload.py

@@ -56,7 +56,7 @@ async def test_apply_channel_feature_action_starts_and_stops_channel(monkeypatch
     monkeypatch.setattr(registry, "discover_channel_names", lambda: ["hot"])
     monkeypatch.setattr(registry, "discover_plugins", lambda enabled_names=None: {})
     monkeypatch.setattr(registry, "discover_enabled", discover_enabled)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda: next(configs))
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda: next(configs))
 
     manager = ChannelManager(disabled, MessageBus())
     manager._started = True
@@ -95,7 +95,7 @@ async def test_apply_channel_feature_action_keeps_running_channel_when_rebuild_f
         "discover_enabled",
         lambda enabled_names, **_kwargs: {"hot": _HotChannel},
     )
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda: enabled)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda: enabled)
 
     manager = ChannelManager(enabled, MessageBus())
     old_channel = manager.channels["hot"]

tests/channels/test_channel_plugins.py

@@ -634,7 +634,7 @@ def test_channels_login_uses_discovered_plugin_class(monkeypatch):
             seen["config"] = self.config
             return True
 
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: Config())
+    monkeypatch.setattr("nanobot.config.loader.load_raw_config", lambda config_path=None: Config())
     monkeypatch.setattr(
         "nanobot.channels.registry.discover_all",
         lambda: {"fakeplugin": _LoginPlugin},
@@ -660,7 +660,7 @@ def test_channels_login_sets_custom_config_path(monkeypatch, tmp_path):
         async def login(self, force: bool = False) -> bool:
             return True
 
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: Config())
+    monkeypatch.setattr("nanobot.config.loader.load_raw_config", lambda config_path=None: Config())
     monkeypatch.setattr(
         "nanobot.config.loader.set_config_path",
         lambda path: seen.__setitem__("config_path", path),
@@ -686,7 +686,7 @@ def test_channels_status_sets_custom_config_path(monkeypatch, tmp_path):
     seen: dict[str, object] = {}
     config_path = tmp_path / "custom-config.json"
 
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: Config())
+    monkeypatch.setattr("nanobot.config.loader.load_raw_config", lambda config_path=None: Config())
     monkeypatch.setattr(
         "nanobot.config.loader.set_config_path",
         lambda path: seen.__setitem__("config_path", path),
@@ -707,7 +707,7 @@ def test_plugins_list_shows_available_features(monkeypatch):
 
     runner = CliRunner()
     config = Config.model_validate({"channels": {"weixin": {"enabled": True}}})
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda config_path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_raw_config", lambda config_path=None: config)
     monkeypatch.setattr("nanobot.channels.registry.discover_channel_names", lambda: ["weixin"])
     monkeypatch.setattr("nanobot.channels.registry.discover_plugins", lambda: {})
     monkeypatch.setattr(

tests/channels/test_websocket_channel.py

@@ -32,7 +32,7 @@ from nanobot.channels.websocket import (
     _parse_inbound_payload,
     publish_runtime_model_update,
 )
-from nanobot.config.loader import load_config, save_config
+from nanobot.config.loader import load_raw_config, save_config
 from nanobot.config.schema import Config, ModelPresetConfig
 from nanobot.session import webui_turns as wth
 from nanobot.session.manager import SessionManager
@@ -2131,7 +2131,7 @@ async def test_settings_api_returns_safe_subset_and_updates_whitelist(
         )
         assert bad_image.status_code == 400
 
-        saved = load_config(config_path)
+        saved = load_raw_config(config_path)
         assert saved.agents.defaults.model == "atomic_chat/test"
         assert saved.agents.defaults.provider == "atomic_chat"
         assert saved.agents.defaults.model_preset == "fast-writing"
@@ -2320,7 +2320,7 @@ def test_update_provider_settings_ignores_api_type_for_non_openai(monkeypatch, t
     })
 
     assert body["providers"]
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
     assert config.providers.custom.api_base == "https://example.test/v1"
     assert config.providers.custom.api_type == "auto"
 

tests/channels/test_websocket_http_routes.py

@@ -1124,7 +1124,7 @@ async def test_channel_configure_route_saves_discord_config_and_hot_reloads(
     ) -> dict[str, Any]:
         assert action == "enable"
         assert query == {"name": ["discord"]}
-        cfg = loader.load_config()
+        cfg = loader.load_raw_config()
         section = dict(getattr(cfg.channels, "discord", {}) or {})
         section["enabled"] = True
         setattr(cfg.channels, "discord", section)
@@ -1151,7 +1151,7 @@ async def test_channel_configure_route_saves_discord_config_and_hot_reloads(
 
     async def channel_feature_action(action: str, name: str) -> dict[str, Any]:
         calls.append((action, name))
-        cfg = loader.load_config()
+        cfg = loader.load_raw_config()
         assert getattr(cfg.channels, "discord")["token"] == "discord-token"
         return {
             "handled": True,

tests/cli/test_commands.py

@@ -224,7 +224,7 @@ def mock_paths():
     """Mock config/workspace paths for test isolation."""
     with patch("nanobot.config.loader.get_config_path") as mock_cp, \
          patch("nanobot.config.loader.save_config") as mock_sc, \
-         patch("nanobot.config.loader.load_config") as mock_lc, \
+         patch("nanobot.config.loader.load_raw_config") as mock_lc, \
          patch("nanobot.cli.commands.get_workspace_path") as mock_ws:
         base_dir = Path("./test_onboard_data")
         if base_dir.exists():
@@ -681,7 +681,7 @@ def test_provider_login_model_implies_set_main_provider(tmp_path):
 def test_provider_login_openai_codex_passes_configured_proxy(monkeypatch):
     proxy = "http://127.0.0.1:23458"
     monkeypatch.setattr(
-        "nanobot.config.loader.load_config",
+        "nanobot.config.loader.load_effective_config",
         lambda: Config.model_validate({"providers": {"openaiCodex": {"proxy": proxy}}}),
     )
 
@@ -706,13 +706,12 @@ def test_provider_login_openai_codex_passes_configured_proxy(monkeypatch):
     assert captured["proxy"] == proxy
 
 
-def test_provider_login_openai_codex_resolves_proxy_env_ref(monkeypatch):
+def test_provider_login_openai_codex_uses_resolved_proxy_from_effective_config(monkeypatch):
     proxy = "http://127.0.0.1:23458"
-    monkeypatch.setenv("CODEX_PROXY_FOR_TEST", proxy)
     monkeypatch.setattr(
-        "nanobot.config.loader.load_config",
+        "nanobot.config.loader.load_effective_config",
         lambda: Config.model_validate(
-            {"providers": {"openaiCodex": {"proxy": "${CODEX_PROXY_FOR_TEST}"}}}
+            {"providers": {"openaiCodex": {"proxy": proxy}}}
         ),
     )
 
@@ -1278,8 +1277,9 @@ def mock_agent_runtime(tmp_path):
     config = Config()
     config.agents.defaults.workspace = str(tmp_path / "default-workspace")
 
-    with patch("nanobot.config.loader.load_config", return_value=config) as mock_load_config, \
-         patch("nanobot.config.loader.resolve_config_env_vars", side_effect=lambda c: c), \
+    with patch(
+        "nanobot.config.loader.load_effective_config", return_value=config
+    ) as mock_load_effective_config, \
          patch("nanobot.cli.commands.sync_workspace_templates") as mock_sync_templates, \
          patch("nanobot.providers.factory.make_provider", return_value=_fake_provider()), \
          patch("nanobot.cli.commands._print_agent_response") as mock_print_response, \
@@ -1296,7 +1296,7 @@ def mock_agent_runtime(tmp_path):
 
         yield {
             "config": config,
-            "load_config": mock_load_config,
+            "load_effective_config": mock_load_effective_config,
             "sync_templates": mock_sync_templates,
             "from_config": mock_from_config,
             "agent_loop": agent_loop,
@@ -1319,7 +1319,7 @@ def test_agent_uses_default_config_when_no_workspace_or_config_flags(mock_agent_
     result = runner.invoke(app, ["agent", "-m", "hello"])
 
     assert result.exit_code == 0
-    assert mock_agent_runtime["load_config"].call_args.args == (None,)
+    assert mock_agent_runtime["load_effective_config"].call_args.args == (None,)
     assert mock_agent_runtime["sync_templates"].call_args.args == (
         mock_agent_runtime["config"].workspace_path,
     )
@@ -1338,7 +1338,9 @@ def test_agent_uses_explicit_config_path(mock_agent_runtime, tmp_path: Path):
     result = runner.invoke(app, ["agent", "-m", "hello", "-c", str(config_path)])
 
     assert result.exit_code == 0
-    assert mock_agent_runtime["load_config"].call_args.args == (config_path.resolve(),)
+    assert mock_agent_runtime["load_effective_config"].call_args.args == (
+        config_path.resolve(),
+    )
 
 
 def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None:
@@ -1353,7 +1355,7 @@ def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None:
         "nanobot.config.loader.set_config_path",
         lambda path: seen.__setitem__("config_path", path),
     )
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
     monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
@@ -1391,7 +1393,7 @@ def test_agent_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path: Pa
     seen: dict[str, Path] = {}
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
     monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
@@ -1440,7 +1442,7 @@ def test_agent_workspace_override_does_not_migrate_legacy_cron(
     seen: dict[str, Path] = {}
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
     monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
@@ -1496,7 +1498,7 @@ def test_agent_custom_config_workspace_does_not_migrate_legacy_cron(
     seen: dict[str, Path] = {}
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
     monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
@@ -1556,7 +1558,9 @@ def test_agent_workspace_override_wins_over_config_workspace(mock_agent_runtime,
     )
 
     assert result.exit_code == 0
-    assert mock_agent_runtime["load_config"].call_args.args == (config_path.resolve(),)
+    assert mock_agent_runtime["load_effective_config"].call_args.args == (
+        config_path.resolve(),
+    )
     assert mock_agent_runtime["config"].agents.defaults.workspace == str(workspace_path)
     assert mock_agent_runtime["sync_templates"].call_args.args == (workspace_path,)
     passed_config = mock_agent_runtime["from_config"].call_args.args[0]
@@ -1675,8 +1679,7 @@ def _patch_cli_command_runtime(
         "nanobot.config.loader.set_config_path",
         set_config_path or (lambda _path: None),
     )
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
-    monkeypatch.setattr("nanobot.config.loader.resolve_config_env_vars", lambda c: c)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr(
         "nanobot.cli.commands.sync_workspace_templates",
         sync_templates or (lambda _path: None),
@@ -2194,7 +2197,7 @@ def test_gateway_unbound_agent_cron_is_skipped(
     seen: dict[str, object] = {}
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
     _patch_gateway_ports_free(monkeypatch)
@@ -2321,7 +2324,7 @@ def test_gateway_bound_cron_runs_as_session_turn(
     seen: dict[str, object] = {"run_records": []}
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
+    monkeypatch.setattr("nanobot.config.loader.load_effective_config", lambda _path=None: config)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
     _patch_gateway_ports_free(monkeypatch)

tests/config/test_config_load_errors.py

@@ -2,12 +2,12 @@ import json
 
 import pytest
 
-from nanobot.config.loader import load_config
+from nanobot.config.loader import load_raw_config
 from nanobot.config.schema import ApiConfig
 
 
 def test_load_config_missing_file_uses_defaults(tmp_path) -> None:
-    config = load_config(tmp_path / "missing.json")
+    config = load_raw_config(tmp_path / "missing.json")
 
     assert config.agents.defaults.model
 
@@ -17,7 +17,15 @@ def test_load_config_invalid_json_fails_fast(tmp_path) -> None:
     config_path.write_text("{broken json", encoding="utf-8")
 
     with pytest.raises(ValueError, match="Failed to load config"):
-        load_config(config_path)
+        load_raw_config(config_path)
+
+
+def test_load_config_rejects_non_object_root(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text("[]", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="config root must be a JSON object"):
+        load_raw_config(config_path)
 
 
 def test_load_config_invalid_schema_fails_fast(tmp_path) -> None:
@@ -28,7 +36,7 @@ def test_load_config_invalid_schema_fails_fast(tmp_path) -> None:
     )
 
     with pytest.raises(ValueError, match="Failed to load config"):
-        load_config(config_path)
+        load_raw_config(config_path)
 
 
 @pytest.mark.parametrize("host", ["0.0.0.0", "::"])

tests/config/test_config_migration.py

@@ -4,7 +4,7 @@ from unittest.mock import patch
 
 import pytest
 
-from nanobot.config.loader import load_config, save_config
+from nanobot.config.loader import apply_config_runtime_policies, load_raw_config, save_config
 from nanobot.security.network import validate_url_target
 
 
@@ -33,7 +33,7 @@ def test_load_config_keeps_max_tokens_and_ignores_legacy_memory_window(tmp_path)
         encoding="utf-8",
     )
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
 
     assert config.agents.defaults.max_tokens == 1234
     assert config.agents.defaults.context_window_tokens == 200_000
@@ -56,7 +56,7 @@ def test_save_config_writes_context_window_tokens_but_not_memory_window(tmp_path
         encoding="utf-8",
     )
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
     save_config(config, config_path)
     saved = json.loads(config_path.read_text(encoding="utf-8"))
     defaults = saved["agents"]["defaults"]
@@ -103,8 +103,8 @@ def test_load_config_warns_and_ignores_legacy_max_messages(tmp_path, field_name)
         encoding="utf-8",
     )
 
-    with patch("nanobot.config.loader.logger.warning") as warning:
-        config = load_config(config_path)
+    with patch("nanobot.config.repository.logger.warning") as warning:
+        config = load_raw_config(config_path)
 
     assert config.agents.defaults.max_tokens == 1234
     assert not hasattr(config.agents.defaults, "max_messages")
@@ -121,8 +121,8 @@ def test_save_config_drops_legacy_max_messages(tmp_path) -> None:
         encoding="utf-8",
     )
 
-    with patch("nanobot.config.loader.logger.warning"):
-        config = load_config(config_path)
+    with patch("nanobot.config.repository.logger.warning"):
+        config = load_raw_config(config_path)
     save_config(config, config_path)
     saved = json.loads(config_path.read_text(encoding="utf-8"))
 
@@ -193,7 +193,7 @@ def test_load_config_migrates_legacy_my_tool_keys(tmp_path) -> None:
         encoding="utf-8",
     )
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
 
     assert config.tools.my.enable is False
     assert config.tools.my.allow_set is True
@@ -213,7 +213,7 @@ def test_save_config_rewrites_legacy_my_tool_keys(tmp_path) -> None:
         encoding="utf-8",
     )
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
     save_config(config, config_path)
     saved = json.loads(config_path.read_text(encoding="utf-8"))
 
@@ -238,13 +238,13 @@ def test_new_my_tool_keys_take_precedence_over_legacy(tmp_path) -> None:
         encoding="utf-8",
     )
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
 
     assert config.tools.my.enable is True
     assert config.tools.my.allow_set is True
 
 
-def test_load_config_resets_ssrf_whitelist_when_next_config_is_empty(tmp_path) -> None:
+def test_runtime_policy_application_resets_ssrf_whitelist(tmp_path) -> None:
     whitelisted = tmp_path / "whitelisted.json"
     whitelisted.write_text(
         json.dumps({"tools": {"ssrfWhitelist": ["100.64.0.0/10"]}}),
@@ -253,12 +253,12 @@ def test_load_config_resets_ssrf_whitelist_when_next_config_is_empty(tmp_path) -
     defaulted = tmp_path / "defaulted.json"
     defaulted.write_text(json.dumps({}), encoding="utf-8")
 
-    load_config(whitelisted)
+    apply_config_runtime_policies(load_raw_config(whitelisted))
     with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
         ok, err = validate_url_target("http://ts.local/api")
         assert ok, err
 
-    load_config(defaulted)
+    apply_config_runtime_policies(load_raw_config(defaulted))
     with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("ts.local", ["100.100.1.1"])):
         ok, _ = validate_url_target("http://ts.local/api")
         assert not ok
@@ -268,7 +268,7 @@ def test_load_config_defaults_local_service_access_to_enabled(tmp_path) -> None:
     config_path = tmp_path / "config.json"
     config_path.write_text(json.dumps({"tools": {}}), encoding="utf-8")
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
 
     assert config.tools.webui_allow_local_service_access is True
 
@@ -280,7 +280,7 @@ def test_load_config_accepts_legacy_local_preview_access(tmp_path) -> None:
         encoding="utf-8",
     )
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
 
     assert config.tools.webui_allow_local_service_access is False
 
@@ -289,7 +289,7 @@ def test_load_config_defaults_remote_package_install_to_disabled(tmp_path) -> No
     config_path = tmp_path / "config.json"
     config_path.write_text(json.dumps({"tools": {}}), encoding="utf-8")
 
-    config = load_config(config_path)
+    config = load_raw_config(config_path)
 
     assert config.tools.webui_allow_remote_package_install is False
 
@@ -306,5 +306,5 @@ def test_load_config_accepts_remote_package_install_aliases(tmp_path) -> None:
         encoding="utf-8",
     )
 
-    assert load_config(camel_path).tools.webui_allow_remote_package_install is True
-    assert load_config(snake_path).tools.webui_allow_remote_package_install is True
+    assert load_raw_config(camel_path).tools.webui_allow_remote_package_install is True
+    assert load_raw_config(snake_path).tools.webui_allow_remote_package_install is True

tests/config/test_env_interpolation.py

@@ -2,54 +2,31 @@ import json
 
 import pytest
 
-from nanobot.config.loader import (
-    _resolve_env_vars,
-    load_config,
-    resolve_config_env_vars,
-    save_config,
-)
+from nanobot.config.loader import load_raw_config, save_config
+from nanobot.config.repository import resolve_config_env_vars
 from nanobot.config.schema import Config
 
 
-class TestResolveEnvVars:
-    def test_replaces_string_value(self, monkeypatch):
-        monkeypatch.setenv("MY_SECRET", "hunter2")
-        assert _resolve_env_vars("${MY_SECRET}") == "hunter2"
+class TestResolveConfig:
+    def test_resolves_multiple_refs_inside_a_value(self, monkeypatch):
+        monkeypatch.setenv("TEST_USER", "alice")
+        monkeypatch.setenv("TEST_PASSWORD", "secret")
+        config = Config.model_validate(
+            {"providers": {"groq": {"apiKey": "${TEST_USER}:${TEST_PASSWORD}"}}}
+        )
 
-    def test_partial_replacement(self, monkeypatch):
-        monkeypatch.setenv("HOST", "example.com")
-        assert _resolve_env_vars("https://${HOST}/api") == "https://example.com/api"
+        resolved = resolve_config_env_vars(config)
 
-    def test_multiple_vars_in_one_string(self, monkeypatch):
-        monkeypatch.setenv("USER", "alice")
-        monkeypatch.setenv("PASS", "secret")
-        assert _resolve_env_vars("${USER}:${PASS}") == "alice:secret"
-
-    def test_nested_dicts(self, monkeypatch):
-        monkeypatch.setenv("TOKEN", "abc123")
-        data = {"channels": {"telegram": {"token": "${TOKEN}"}}}
-        result = _resolve_env_vars(data)
-        assert result["channels"]["telegram"]["token"] == "abc123"
-
-    def test_lists(self, monkeypatch):
-        monkeypatch.setenv("VAL", "x")
-        assert _resolve_env_vars(["${VAL}", "plain"]) == ["x", "plain"]
-
-    def test_ignores_non_strings(self):
-        assert _resolve_env_vars(42) == 42
-        assert _resolve_env_vars(True) is True
-        assert _resolve_env_vars(None) is None
-        assert _resolve_env_vars(3.14) == 3.14
-
-    def test_plain_strings_unchanged(self):
-        assert _resolve_env_vars("no vars here") == "no vars here"
+        assert resolved.providers.groq.api_key == "alice:secret"
 
     def test_missing_var_raises(self):
+        config = Config.model_validate(
+            {"providers": {"groq": {"apiKey": "${DOES_NOT_EXIST}"}}}
+        )
+
         with pytest.raises(ValueError, match="DOES_NOT_EXIST"):
-            _resolve_env_vars("${DOES_NOT_EXIST}")
+            resolve_config_env_vars(config)
 
-
-class TestResolveConfig:
     def test_resolves_env_vars_in_config(self, tmp_path, monkeypatch):
         monkeypatch.setenv("TEST_API_KEY", "resolved-key")
         config_path = tmp_path / "config.json"
@@ -60,7 +37,7 @@ class TestResolveConfig:
             encoding="utf-8",
         )
 
-        raw = load_config(config_path)
+        raw = load_raw_config(config_path)
         assert raw.providers.groq.api_key == "${TEST_API_KEY}"
 
         resolved = resolve_config_env_vars(raw)
@@ -76,7 +53,7 @@ class TestResolveConfig:
             encoding="utf-8",
         )
 
-        raw = load_config(config_path)
+        raw = load_raw_config(config_path)
         save_config(raw, config_path)
 
         saved = json.loads(config_path.read_text(encoding="utf-8"))
@@ -91,14 +68,14 @@ class TestResolveConfig:
             encoding="utf-8",
         )
 
-        config = load_config(config_path)
+        config = load_raw_config(config_path)
         config.agents.defaults.max_tokens = 1234
         save_config(config, config_path)
 
         saved = json.loads(config_path.read_text(encoding="utf-8"))
         assert saved["agents"]["defaults"]["dream"]["cron"] == "0 */4 * * *"
 
-        reloaded = load_config(config_path)
+        reloaded = load_raw_config(config_path)
         schedule = reloaded.agents.defaults.dream.build_schedule("UTC")
         assert schedule.kind == "cron"
         assert schedule.expr == "0 */4 * * *"
@@ -119,7 +96,7 @@ class TestResolveConfig:
             encoding="utf-8",
         )
 
-        config = load_config(config_path)
+        config = load_raw_config(config_path)
         save_config(config, config_path)
 
         saved = json.loads(config_path.read_text(encoding="utf-8"))
@@ -149,7 +126,7 @@ class TestResolveConfig:
         assert saved["providers"]["openaiCodex"] == {"proxy": proxy}
         assert saved["providers"]["groq"]["apiKey"] == "groq-secret"
 
-        reloaded = load_config(config_path)
+        reloaded = load_raw_config(config_path)
         assert reloaded.providers.openai_codex.proxy == proxy
         assert reloaded.providers.openai_codex.api_key is None
 
@@ -166,7 +143,7 @@ class TestResolveConfig:
             encoding="utf-8",
         )
 
-        raw = load_config(config_path)
+        raw = load_raw_config(config_path)
         assert raw.providers.openai_codex.api_key == "secret"
 
         resolved = resolve_config_env_vars(raw)
@@ -190,7 +167,7 @@ class TestResolveConfig:
             encoding="utf-8",
         )
 
-        raw = load_config(config_path)
+        raw = load_raw_config(config_path)
         resolved = resolve_config_env_vars(raw)
 
         assert resolved.providers.groq.api_key == "resolved-key"

tests/config/test_repository.py

@@ -0,0 +1,235 @@
+import json
+import os
+import socket
+import stat
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from nanobot.config.loader import load_raw_config
+from nanobot.config.repository import ConfigConflictError, FileConfigRepository
+from nanobot.security.network import configure_ssrf_whitelist, validate_url_target
+
+
+def _fake_resolve(host: str, results: list[str]):
+    def _resolver(hostname, port, family=0, type_=0):
+        if hostname == host:
+            return [
+                (socket.AF_INET, socket.SOCK_STREAM, 0, "", (ip, 0))
+                for ip in results
+            ]
+        raise socket.gaierror(f"cannot resolve {hostname}")
+
+    return _resolver
+
+
+def test_public_api_exposes_explicit_loaders_without_legacy_alias() -> None:
+    import nanobot.config as config_api
+    from nanobot.config import loader
+
+    assert config_api.load_raw_config is loader.load_raw_config
+    assert config_api.load_effective_config is loader.load_effective_config
+    assert not hasattr(config_api, "load_config")
+    assert not hasattr(loader, "load_config")
+
+
+def test_raw_and_effective_snapshots_keep_secret_templates_separate(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    path = tmp_path / "config.json"
+    path.write_text(
+        json.dumps({"providers": {"groq": {"apiKey": "${GROQ_TOKEN}"}}}),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("GROQ_TOKEN", "resolved-secret")
+    repository = FileConfigRepository(path)
+
+    raw = repository.load_raw()
+    effective = repository.load_effective()
+
+    assert raw.config.providers.groq.api_key == "${GROQ_TOKEN}"
+    assert effective.config.providers.groq.api_key == "resolved-secret"
+    assert effective.config is not raw.config
+    assert effective.revision == raw.revision
+
+
+def test_update_uses_latest_raw_config_and_reports_changed_paths(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    path = tmp_path / "config.json"
+    path.write_text(
+        json.dumps({"providers": {"groq": {"apiKey": "${GROQ_TOKEN}"}}}),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("GROQ_TOKEN", "resolved-secret")
+    repository = FileConfigRepository(path)
+
+    commit = repository.update(
+        lambda config: setattr(config.agents.defaults, "timezone", "Asia/Shanghai")
+    )
+
+    saved = json.loads(path.read_text(encoding="utf-8"))
+    assert saved["providers"]["groq"]["apiKey"] == "${GROQ_TOKEN}"
+    assert saved["agents"]["defaults"]["timezone"] == "Asia/Shanghai"
+    assert commit.before.revision != commit.after.revision
+    assert commit.changed_paths == frozenset({"agents.defaults.timezone"})
+
+
+def test_update_rejects_stale_expected_revision(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    repository = FileConfigRepository(path)
+    initial = repository.load_raw()
+    repository.update(lambda config: setattr(config.api, "port", 9001))
+
+    with pytest.raises(ConfigConflictError, match="Config changed"):
+        repository.update(
+            lambda config: setattr(config.api, "port", 9002),
+            expected_revision=initial.revision,
+        )
+
+    assert repository.load_raw().config.api.port == 9001
+
+
+def test_repositories_for_same_path_serialize_threaded_updates(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    first = FileConfigRepository(path)
+    second = FileConfigRepository(path)
+    first_mutator_entered = threading.Event()
+    release_first = threading.Event()
+    second_mutator_entered = threading.Event()
+
+    def update_first() -> None:
+        def mutate(config):
+            first_mutator_entered.set()
+            assert release_first.wait(timeout=5)
+            config.api.port = 9001
+
+        first.update(mutate)
+
+    def update_second() -> None:
+        def mutate(config):
+            second_mutator_entered.set()
+            config.agents.defaults.timezone = "Asia/Shanghai"
+
+        second.update(mutate)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        first_future = pool.submit(update_first)
+        assert first_mutator_entered.wait(timeout=5)
+        second_future = pool.submit(update_second)
+        assert not second_mutator_entered.wait(timeout=0.1)
+        release_first.set()
+        first_future.result(timeout=5)
+        second_future.result(timeout=5)
+
+    config = first.load_raw().config
+    assert config.api.port == 9001
+    assert config.agents.defaults.timezone == "Asia/Shanghai"
+
+
+def test_repositories_for_different_paths_are_isolated(tmp_path: Path) -> None:
+    first_path = tmp_path / "first.json"
+    second_path = tmp_path / "second.json"
+    first = FileConfigRepository(first_path)
+    second = FileConfigRepository(second_path)
+
+    first.update(lambda config: setattr(config.api, "port", 9001))
+    second.update(lambda config: setattr(config.api, "port", 9002))
+
+    assert first.load_raw().config.api.port == 9001
+    assert second.load_raw().config.api.port == 9002
+    assert json.loads(first_path.read_text(encoding="utf-8"))["api"]["port"] == 9001
+    assert json.loads(second_path.read_text(encoding="utf-8"))["api"]["port"] == 9002
+
+
+def test_noop_update_keeps_revision_and_does_not_rewrite(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    repository = FileConfigRepository(path)
+    repository.update(lambda config: setattr(config.api, "port", 9001))
+    before = repository.load_raw()
+
+    commit = repository.update(lambda config: setattr(config.api, "port", 9001))
+
+    assert commit.changed_paths == frozenset()
+    assert commit.before.revision == before.revision
+    assert commit.after.revision == before.revision
+
+
+def test_atomic_save_keeps_previous_file_when_replace_fails(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    path.write_text('{"api": {"port": 9000}}', encoding="utf-8")
+    repository = FileConfigRepository(path)
+
+    with patch("nanobot.config.repository.os.replace", side_effect=OSError("replace failed")):
+        with pytest.raises(OSError, match="replace failed"):
+            repository.update(lambda config: setattr(config.api, "port", 9001))
+
+    assert json.loads(path.read_text(encoding="utf-8"))["api"]["port"] == 9000
+    assert list(tmp_path.glob(".config.json.*.tmp")) == []
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Windows does not expose POSIX file modes")
+def test_atomic_save_preserves_existing_file_mode(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    path.write_text('{"api": {"port": 9000}}', encoding="utf-8")
+    path.chmod(0o600)
+
+    FileConfigRepository(path).update(lambda config: setattr(config.api, "port", 9001))
+
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+
+
+def test_atomic_save_restores_mode_before_writing(tmp_path: Path, monkeypatch) -> None:
+    path = tmp_path / "config.json"
+    path.write_text('{"api": {"port": 9000}}', encoding="utf-8")
+    path.chmod(0o600)
+    observed_sizes: list[int] = []
+    real_chmod = os.chmod
+
+    def capture_size_before_chmod(target: str | bytes | Path, mode: int) -> None:
+        observed_sizes.append(Path(target).stat().st_size)
+        real_chmod(target, mode)
+
+    monkeypatch.setattr("nanobot.config.repository.os.chmod", capture_size_before_chmod)
+
+    FileConfigRepository(path).update(lambda config: setattr(config.api, "port", 9001))
+
+    assert observed_sizes == [0]
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Windows handles chmod differently")
+def test_atomic_save_keeps_previous_file_when_mode_restore_fails(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    path.write_text('{"api": {"port": 9000}}', encoding="utf-8")
+    path.chmod(0o600)
+
+    with patch("nanobot.config.repository.os.chmod", side_effect=OSError("chmod failed")):
+        with pytest.raises(OSError, match="chmod failed"):
+            FileConfigRepository(path).update(
+                lambda config: setattr(config.api, "port", 9001)
+            )
+
+    assert json.loads(path.read_text(encoding="utf-8"))["api"]["port"] == 9000
+    assert list(tmp_path.glob(".config.json.*.tmp")) == []
+
+
+def test_loading_config_does_not_change_process_network_policy(tmp_path: Path) -> None:
+    path = tmp_path / "config.json"
+    path.write_text(json.dumps({"tools": {"ssrfWhitelist": []}}), encoding="utf-8")
+    configure_ssrf_whitelist(["100.64.0.0/10"])
+    try:
+        load_raw_config(path)
+
+        with patch(
+            "nanobot.security.network.socket.getaddrinfo",
+            _fake_resolve("ts.local", ["100.100.1.1"]),
+        ):
+            ok, error = validate_url_target("http://ts.local/api")
+            assert ok, error
+    finally:
+        configure_ssrf_whitelist([])

tests/test_nanobot_facade.py

@@ -120,7 +120,7 @@ def test_from_config_rejects_multiple_model_selectors(tmp_path):
 def test_from_config_default_path():
     from nanobot.config.schema import Config
 
-    with patch("nanobot.config.loader.load_config") as mock_load, \
+    with patch("nanobot.config.loader.load_effective_config") as mock_load, \
          patch("nanobot.providers.factory.make_provider") as mock_prov:
         mock_load.return_value = Config()
         mock_prov.return_value = MagicMock()

tests/webui/test_channel_validation.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import httpx
 import pytest
 
-from nanobot.config.loader import load_config, save_config
+from nanobot.config.loader import load_raw_config, save_config
 from nanobot.config.schema import Config
 from nanobot.webui import channel_validation
 from nanobot.webui.channel_validation import validate_channel_config
@@ -35,7 +35,7 @@ def test_validate_channel_does_not_write_config(tmp_path, monkeypatch: pytest.Mo
     )
 
     assert payload["status"] == "connected"
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.channels.slack["appToken"] == "xapp-old"
     assert saved.channels.slack["botToken"] == "xoxb-old"
 
@@ -101,7 +101,7 @@ def test_validate_email_presets_are_checked_without_saving(
 
     assert payload["status"] == "connected"
     assert payload["can_enable"] is True
-    assert not hasattr(load_config(config_path).channels, "email")
+    assert not hasattr(load_raw_config(config_path).channels, "email")
 
 
 def test_validate_email_blocks_private_targets_when_local_access_is_disabled(

tests/webui/test_mcp_presets_api.py

@@ -4,7 +4,7 @@ import asyncio
 
 import pytest
 
-from nanobot.config.loader import load_config
+from nanobot.config.loader import load_raw_config
 from nanobot.webui.mcp_presets_api import (
     McpPresetError,
     custom_mcp_action,
@@ -76,7 +76,7 @@ def test_enable_browserbase_writes_scrubbed_config_payload(
     assert preset["installed"] is True
     assert preset["configured"] is True
     assert "bb_live_secret" not in str(payload)
-    config = load_config()
+    config = load_raw_config()
     assert "browserbaseApiKey=bb_live_secret" in config.tools.mcp_servers["browserbase"].url
 
 
@@ -107,7 +107,7 @@ def test_enable_context7_optional_api_key_appends_arg(
     assert "ctx7_secret" not in str(payload)
     row = next(item for item in payload["presets"] if item["name"] == "context7")
     assert row["configured"] is True
-    config = load_config()
+    config = load_raw_config()
     assert config.tools.mcp_servers["context7"].args == [
         "-y",
         "@upstash/context7-mcp@latest",
@@ -124,7 +124,7 @@ def test_enable_stdio_preset_uses_config_scoped_cwd(
 
     mcp_presets_action("enable", {"name": ["playwright"]})
 
-    config = load_config()
+    config = load_raw_config()
     cwd = config.tools.mcp_servers["playwright"].cwd
     assert cwd == str(tmp_path / "mcp" / "playwright")
     assert (tmp_path / "mcp" / "playwright").is_dir()
@@ -137,7 +137,7 @@ def test_enable_no_auth_remote_presets_write_url(tmp_path, monkeypatch: pytest.M
     mcp_presets_action("enable", {"name": ["exa"]})
     mcp_presets_action("enable", {"name": ["firecrawl"]})
 
-    config = load_config()
+    config = load_raw_config()
     assert config.tools.mcp_servers["microsoft-learn"].url == "https://learn.microsoft.com/api/mcp"
     assert config.tools.mcp_servers["exa"].url == "https://mcp.exa.ai/mcp"
     assert config.tools.mcp_servers["firecrawl"].url == "https://mcp.firecrawl.dev/v2/mcp"
@@ -154,7 +154,7 @@ def test_firecrawl_preset_is_keyless(tmp_path, monkeypatch: pytest.MonkeyPatch)
     assert row["required_fields"] == []
     assert row["configured"] is True
     assert "Keyless" in row["note"]
-    config = load_config()
+    config = load_raw_config()
     assert config.tools.mcp_servers["firecrawl"].type == "streamableHttp"
     assert config.tools.mcp_servers["firecrawl"].url == "https://mcp.firecrawl.dev/v2/mcp"
     assert config.tools.mcp_servers["firecrawl"].env == {}
@@ -173,7 +173,7 @@ def test_remove_mcp_preset_updates_config(tmp_path, monkeypatch: pytest.MonkeyPa
     assert payload["last_action"]["removed"] is True
     assert payload["last_action"]["managed_paths_removed"] == ["runtime:mcp/playwright"]
     assert not managed_cwd.exists()
-    config = load_config()
+    config = load_raw_config()
     assert "playwright" not in config.tools.mcp_servers
 
 
@@ -196,7 +196,7 @@ def test_remove_custom_mcp_server_preserves_user_cwd(tmp_path, monkeypatch: pyte
 
     assert payload["last_action"]["ok"] is True
     assert user_cwd.exists()
-    config = load_config()
+    config = load_raw_config()
     assert "internal-docs" not in config.tools.mcp_servers
 
 
@@ -330,7 +330,7 @@ def test_custom_mcp_server_writes_config_and_catalog_row(
     assert row["manifest"]["capabilities"][0]["command"] == "node"
     assert "server.js" not in str(row["manifest"])
     assert "docs-secret-value" not in str(payload)
-    config = load_config()
+    config = load_raw_config()
     assert config.tools.mcp_servers["internal-docs"].args == ["server.js"]
     assert config.tools.mcp_servers["internal-docs"].env["DOCS_TOKEN"] == "docs-secret-value"
 
@@ -356,7 +356,7 @@ def test_import_mcp_config_and_tool_allowlist(
     )
 
     assert payload["last_action"]["message"] == "Imported 2 MCP server(s)."
-    config = load_config()
+    config = load_raw_config()
     assert config.tools.mcp_servers["docs"].command == "npx"
     assert config.tools.mcp_servers["docs"].args == ["-y", "docs-mcp"]
     assert config.tools.mcp_servers["remote-docs"].type == "sse"
@@ -374,7 +374,7 @@ def test_import_mcp_config_and_tool_allowlist(
 
     row = next(item for item in payload["presets"] if item["name"] == "docs")
     assert row["enabled_tools"] == ["mcp_docs_search"]
-    assert load_config().tools.mcp_servers["docs"].enabled_tools == ["mcp_docs_search"]
+    assert load_raw_config().tools.mcp_servers["docs"].enabled_tools == ["mcp_docs_search"]
 
     payload = custom_mcp_action(
         "tools",
@@ -386,7 +386,7 @@ def test_import_mcp_config_and_tool_allowlist(
 
     row = next(item for item in payload["presets"] if item["name"] == "docs")
     assert row["enabled_tools"] == []
-    assert load_config().tools.mcp_servers["docs"].enabled_tools == []
+    assert load_raw_config().tools.mcp_servers["docs"].enabled_tools == []
 
 
 def test_normalize_mcp_preset_mentions_accepts_configured_custom_server(

tests/webui/test_settings_api.py

@@ -7,7 +7,7 @@ from types import SimpleNamespace
 import httpx
 import pytest
 
-from nanobot.config.loader import load_config, save_config
+from nanobot.config.loader import load_raw_config, save_config
 from nanobot.config.schema import Config, ModelPresetConfig
 from nanobot.providers.registry import find_by_name
 from nanobot.webui.settings_api import (
@@ -95,7 +95,7 @@ def test_update_api_settings_requires_key_for_network_access(
         "port": ["9900"],
         "api_key": ["secret-token"],
     })
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.api.host == "0.0.0.0"
     assert saved.api.port == 9900
     assert saved.api.api_key == "secret-token"
@@ -124,7 +124,7 @@ def test_update_api_settings_allows_alternate_loopback_without_key(
 
     update_api_settings({"host": ["127.0.0.2"], "port": ["8900"]})
 
-    assert load_config(config_path).api.host == "127.0.0.2"
+    assert load_raw_config(config_path).api.host == "127.0.0.2"
 
 
 def _dynamic_provider_config(
@@ -174,7 +174,7 @@ def test_create_model_configuration_writes_label_and_selects(
     rows = {row["name"]: row for row in payload["model_presets"]}
     assert rows["fast-writing"]["label"] == "Fast writing"
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.agents.defaults.model_preset == "fast-writing"
     assert saved.model_presets["fast-writing"].label == "Fast writing"
     assert saved.model_presets["fast-writing"].model == "openai/gpt-4.1-mini"
@@ -209,7 +209,7 @@ def test_create_model_configuration_accepts_dynamic_custom_provider(
 
     assert payload["agent"]["model_preset"] == "tenant-model"
     assert payload["agent"]["provider"] == DYNAMIC_PROVIDER_NAME
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.model_presets["tenant-model"].provider == DYNAMIC_PROVIDER_NAME
     assert saved.model_presets["tenant-model"].model == "gpt-4o-mini"
 
@@ -292,7 +292,7 @@ def test_update_model_configuration_edits_named_preset_and_selects(
 
     assert payload["agent"]["model_preset"] == "codex"
     assert payload["agent"]["model"] == "openai-codex/gpt-5.5"
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.agents.defaults.model_preset == "codex"
     assert saved.model_presets["codex"].label == "Codex"
     assert saved.model_presets["codex"].provider == "openai_codex"
@@ -318,7 +318,7 @@ def test_update_provider_settings_updates_dynamic_custom_provider(
     providers = {row["name"]: row for row in payload["providers"]}
     assert providers[DYNAMIC_PROVIDER_NAME]["api_base"] == "https://new.example/v1"
     assert providers[DYNAMIC_PROVIDER_NAME]["api_key_hint"] == "••••"
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     dynamic_provider = saved.providers.model_extra[DYNAMIC_PROVIDER_NAME]
     assert dynamic_provider.api_base == "https://new.example/v1"
     assert dynamic_provider.api_key == "sk-test"
@@ -336,7 +336,7 @@ def test_update_agent_settings_accepts_context_window_options(
     payload = update_agent_settings({"context_window_tokens": ["200000"]})
 
     assert payload["agent"]["context_window_tokens"] == 200000
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.agents.defaults.context_window_tokens == 200000
 
 
@@ -362,7 +362,7 @@ def test_update_model_configuration_accepts_context_window_options(
     )
 
     assert payload["agent"]["context_window_tokens"] == 262144
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.model_presets["codex"].context_window_tokens == 262144
 
 
@@ -555,7 +555,7 @@ def test_update_web_search_settings_accepts_keenable_without_api_key(
 
     payload = update_web_search_settings({"provider": ["keenable"]})
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.tools.web.search.provider == "keenable"
     assert saved.tools.web.search.api_key == ""
     option = next(item for item in payload["web_search"]["providers"] if item["name"] == "keenable")
@@ -575,7 +575,7 @@ def test_update_web_search_settings_can_clear_optional_api_key(
 
     update_web_search_settings({"provider": ["keenable"], "api_key": [""]})
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.tools.web.search.provider == "keenable"
     assert saved.tools.web.search.api_key == ""
 
@@ -720,7 +720,7 @@ def test_update_transcription_settings_writes_top_level_only(
         }
     )
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.channels.transcription_provider == "openai"
     assert saved.channels.transcription_language == "en"
     assert saved.transcription.enabled is True
@@ -750,7 +750,7 @@ def test_update_transcription_settings_accepts_openrouter(
         }
     )
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.transcription.provider == "openrouter"
     assert saved.transcription.model == "nvidia/parakeet-tdt-0.6b-v3"
     assert payload["transcription"]["provider"] == "openrouter"
@@ -775,7 +775,7 @@ def test_update_transcription_settings_accepts_xiaomi_mimo(
         }
     )
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.transcription.provider == "xiaomi_mimo"
     assert saved.transcription.model == "mimo-v2.5-asr"
     assert saved.transcription.language == "zh"
@@ -800,7 +800,7 @@ def test_update_transcription_settings_accepts_assemblyai(
         }
     )
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.transcription.provider == "assemblyai"
     assert saved.transcription.model == "universal-3-pro"
     assert payload["transcription"]["provider"] == "assemblyai"
@@ -881,7 +881,7 @@ def test_update_network_safety_settings_writes_local_service_flag(
         }
     )
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     saved_raw = json.loads(config_path.read_text(encoding="utf-8"))
     assert saved.tools.webui_allow_local_service_access is False
     assert saved_raw["tools"]["webuiAllowLocalServiceAccess"] is False
@@ -917,7 +917,7 @@ def test_update_network_safety_settings_default_access_is_webui_only(
 
     payload = update_network_safety_settings({"webui_default_access_mode": ["full"]})
 
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert config_path.read_text(encoding="utf-8") == before
     assert saved.tools.restrict_to_workspace is False
     assert payload["advanced"]["webui_default_access_mode"] == "full"
@@ -1225,7 +1225,7 @@ def test_create_model_configuration_accepts_configured_oauth_provider(
     )
 
     assert payload["agent"]["model_preset"] == "codex"
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.model_presets["codex"].provider == "openai_codex"
 
 
@@ -1312,7 +1312,7 @@ def test_create_model_configuration_accepts_azure_openai_aad_mode(
     )
 
     assert payload["agent"]["model_preset"] == "azure-aad"
-    saved = load_config(config_path)
+    saved = load_raw_config(config_path)
     assert saved.model_presets["azure-aad"].provider == "azure_openai"
     assert saved.model_presets["azure-aad"].model == "my-deployment"