diff --git a/reflexio/lib/_config.py b/reflexio/lib/_config.py
index c0cf0bcdd..34b6641ea 100644
--- a/reflexio/lib/_config.py
+++ b/reflexio/lib/_config.py
@@ -1,12 +1,19 @@
 from typing import Any
 
 from reflexio.lib._base import ReflexioBase
-from reflexio.models.api_schema.retriever_schema import SetConfigResponse
+from reflexio.models.api_schema.retriever_schema import (
+    CONFIG_UNCHANGED_MESSAGE,
+    SetConfigResponse,
+)
 from reflexio.models.config_schema import Config, StorageConfigManagedSupabase
+from reflexio.server.services.configurator.base_configurator import PreparedConfigWrite
+from reflexio.server.services.configurator.config_storage import ConfigWriteConflict
 
 
 class ConfigMixin(ReflexioBase):
-    def set_config(self, config: Config | dict) -> SetConfigResponse:
+    def set_config(
+        self, config: Config | dict | PreparedConfigWrite
+    ) -> SetConfigResponse:
         """Set configuration for the organization.
 
         Args:
@@ -17,9 +24,12 @@ def set_config(self, config: Config | dict) -> SetConfigResponse:
         """
         try:
             configurator = self.request_context.configurator
-            if isinstance(config, dict):
-                config = configurator.normalize_config_payload(config)
-                config = Config(**config)
+            prepared = (
+                config
+                if isinstance(config, PreparedConfigWrite)
+                else configurator.prepare_config_write(config)
+            )
+            validated_config = prepared.config
 
             # Validate storage connection before setting config.
             # If no storage_config provided, or the caller round-tripped the
@@ -30,12 +40,14 @@ def set_config(self, config: Config | dict) -> SetConfigResponse:
             # an unfillable StorageConfigManagedSupabase and fails with
             # "Storage configuration is incomplete".
             current_storage_config = configurator.get_current_storage_configuration()
-            storage_config = config.storage_config
+            storage_config = validated_config.storage_config
             if storage_config is None or isinstance(
                 storage_config, StorageConfigManagedSupabase
             ):
                 storage_config = current_storage_config
-                config.storage_config = storage_config
+                validated_config.storage_config = storage_config
+                if "storage_config" in prepared.payload:
+                    prepared.payload["storage_config"] = storage_config
 
             storage_config_changed = storage_config != current_storage_config
             if storage_config_changed or current_storage_config is None:
@@ -62,9 +74,18 @@ def set_config(self, config: Config | dict) -> SetConfigResponse:
                     )
 
             # Only set config if validation passed
-            configurator.set_config(config)
+            changed = configurator.commit_config_write(prepared)
 
-            return SetConfigResponse(success=True, msg="Configuration set successfully")
+            return SetConfigResponse(
+                success=True,
+                msg=(
+                    "Configuration set successfully"
+                    if changed
+                    else CONFIG_UNCHANGED_MESSAGE
+                ),
+            )
+        except ConfigWriteConflict:
+            raise
         except Exception as e:
             return SetConfigResponse(
                 success=False, msg=f"Failed to set configuration: {str(e)}"
diff --git a/reflexio/models/api_schema/retriever_schema.py b/reflexio/models/api_schema/retriever_schema.py
index b22cd1617..a714eac4e 100644
--- a/reflexio/models/api_schema/retriever_schema.py
+++ b/reflexio/models/api_schema/retriever_schema.py
@@ -222,6 +222,12 @@ class GetProfileStatisticsResponse(BaseModel):
     msg: str | None = None
 
 
+# Sentinel message for a successful config write that persisted no change.
+# HTTP routes key cache invalidation off this exact value, so producers and
+# consumers must share the constant instead of retyping the string.
+CONFIG_UNCHANGED_MESSAGE = "Configuration unchanged"
+
+
 class SetConfigResponse(BaseModel):
     success: bool
     msg: str | None = None
diff --git a/reflexio/server/README.md b/reflexio/server/README.md
index 808f7f0e6..4b9e938f5 100644
--- a/reflexio/server/README.md
+++ b/reflexio/server/README.md
@@ -101,6 +101,13 @@ Description: FastAPI backend server that processes user interactions to generate
 
 **Pattern**: Core route handlers call `Reflexio` through `get_reflexio(org_id)`; endpoint helper files should not instantiate `Reflexio` directly.
 
+Config writes validate the complete resulting document before persistence.
+`set_config` replaces the shared document, while `update_config` applies a
+top-level shallow patch that is re-merged against the latest payload under the
+storage lock. Unknown or malformed fields return HTTP 422. Concurrent writers
+fail immediately with HTTP 409 (`Configuration update already in progress`);
+successful response shapes are unchanged.
+
 ## Extension Registry
 
 **File**: `extensions.py`
diff --git a/reflexio/server/routes/config.py b/reflexio/server/routes/config.py
index d02362792..aa6c07493 100644
--- a/reflexio/server/routes/config.py
+++ b/reflexio/server/routes/config.py
@@ -15,14 +15,12 @@
 )
 
 from reflexio.models.api_schema.retriever_schema import (
+    CONFIG_UNCHANGED_MESSAGE,
     SetConfigResponse,
 )
 from reflexio.models.api_schema.service_schemas import (
     MyConfigResponse,
 )
-from reflexio.models.config_schema import (
-    Config,
-)
 from reflexio.server.api_endpoints import (
     account_api,
 )
@@ -31,11 +29,19 @@
 )
 from reflexio.server.cache import reflexio_cache
 from reflexio.server.rate_limit import limiter
+from reflexio.server.services.configurator.config_storage import ConfigWriteConflict
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def _config_write_conflict() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_409_CONFLICT,
+        detail="Configuration update already in progress",
+    )
+
+
 @router.get(
     "/api/my_config",
     response_model=MyConfigResponse,
@@ -93,10 +99,7 @@ def set_config(
     reflexio = reflexio_cache.get_reflexio(org_id=org_id)
     configurator = reflexio.request_context.configurator
     try:
-        normalized_config = configurator.normalize_config_payload(config)
-        if not isinstance(normalized_config, dict):
-            normalized_config = config
-        Config.model_validate(normalized_config)
+        prepared = configurator.prepare_config_write(config)
     except ValidationError as exc:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
@@ -104,10 +107,13 @@ def set_config(
         ) from exc
 
     # Set the config using Reflexio's set_config method
-    response = reflexio.set_config(normalized_config)
+    try:
+        response = reflexio.set_config(prepared)
+    except ConfigWriteConflict as exc:
+        raise _config_write_conflict() from exc
 
     # Invalidate cache on successful config change to ensure fresh instance next request
-    if response.success:
+    if response.success and response.msg != CONFIG_UNCHANGED_MESSAGE:
         reflexio_cache.invalidate_reflexio_cache(org_id=org_id)
 
     return response
@@ -156,9 +162,7 @@ def update_config(
     from pydantic import ValidationError
 
     reflexio = reflexio_cache.get_reflexio(org_id=org_id)
-    existing_config = reflexio.request_context.configurator.get_config()
-    existing = existing_config.model_dump(mode="python")
-    merged = {**existing, **partial}
+    configurator = reflexio.request_context.configurator
     # Pydantic validates the merged shape and rejects unknown / malformed
     # fields here, before storage validation in reflexio.set_config.
     # Convert ValidationError into 422 so callers passing a partial that
@@ -166,7 +170,7 @@ def update_config(
     # {"user_playbook_extractor_config": {"aggregation_config": {...}}})
     # get a clean client-error response instead of a 500.
     try:
-        merged_config = Config(**merged)
+        prepared = configurator.prepare_config_write(partial, partial=True)
     except ValidationError as exc:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
@@ -182,12 +186,11 @@ def update_config(
                 "validation_errors": exc.errors(),
             },
         ) from exc
-    if merged_config.model_dump(mode="python") == existing:
-        logger.info("Skipping no-op config update for org %s", org_id)
-        return SetConfigResponse(success=True, msg="Configuration unchanged")
-
-    response = reflexio.set_config(merged_config)
-    if response.success:
+    try:
+        response = reflexio.set_config(prepared)
+    except ConfigWriteConflict as exc:
+        raise _config_write_conflict() from exc
+    if response.success and response.msg != CONFIG_UNCHANGED_MESSAGE:
         reflexio_cache.invalidate_reflexio_cache(org_id=org_id)
     return response
 
diff --git a/reflexio/server/services/configurator/base_configurator.py b/reflexio/server/services/configurator/base_configurator.py
index bddfd94d6..0b0dc3169 100644
--- a/reflexio/server/services/configurator/base_configurator.py
+++ b/reflexio/server/services/configurator/base_configurator.py
@@ -3,12 +3,18 @@
 import logging
 from abc import ABC, abstractmethod
 from collections.abc import Callable
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, ClassVar
 
 from pydantic import BaseModel
 
-from reflexio.models.config_schema import Config, StorageConfig, StorageConfigTest
+from reflexio.models.config_schema import (
+    Config,
+    StorageConfig,
+    StorageConfigTest,
+    validate_stored_config,
+)
 from reflexio.server.services.configurator.config_storage import ConfigStorage
 from reflexio.server.services.storage.error import StorageError
 from reflexio.server.services.storage.storage_base import BaseStorage
@@ -23,6 +29,15 @@
 }
 
 
+@dataclass(frozen=True)
+class PreparedConfigWrite:
+    """Side-effect-free, validated config write ready for storage checks."""
+
+    config: Config
+    payload: dict[str, Any]
+    partial: bool
+
+
 class BaseConfigurator(ABC):
     """Abstract base for organization configurators.
 
@@ -82,6 +97,48 @@ def normalize_config_payload(self, config: dict[str, Any]) -> dict[str, Any]:
         """Normalize raw API config payloads before Pydantic validation."""
         return config
 
+    def prepare_config_write(
+        self,
+        config: Config | dict[str, Any],
+        *,
+        partial: bool = False,
+    ) -> PreparedConfigWrite:
+        """Validate a full or partial write without mutating configurator state."""
+        if isinstance(config, Config):
+            payload = config.model_dump(mode="python")
+        else:
+            payload = self.normalize_config_payload(dict(config))
+
+        current = self.config.model_dump(mode="python")
+        candidate = {**current, **payload} if partial else payload
+        validated = Config.model_validate(candidate)
+        return PreparedConfigWrite(
+            config=validated,
+            payload=dict(payload),
+            partial=partial,
+        )
+
+    def commit_config_write(self, prepared: PreparedConfigWrite) -> bool:
+        """Persist a prepared write and update in-memory state.
+
+        Returns:
+            bool: Whether the persisted payload changed.
+        """
+
+        def replace_payload(current: dict[str, Any]) -> dict[str, Any]:
+            if not prepared.partial:
+                return prepared.config.model_dump(mode="json")
+            current_config = validate_stored_config(current)
+            candidate = {
+                **current_config.model_dump(mode="python"),
+                **prepared.payload,
+            }
+            return Config.model_validate(candidate).model_dump(mode="json")
+
+        result = self.config_storage.update_config_payload(replace_payload)
+        self.config = Config.model_validate(result.payload)
+        return result.changed
+
     def get_prompt_bank_paths(self) -> list[Path]:
         """Return additional prompt banks this configurator contributes."""
         return []
@@ -93,8 +150,8 @@ def get_agent_context(self) -> str:
         return context.strip()
 
     def set_config(self, config: Config) -> None:
-        self.config_storage.save_config(config=config)
-        self.config = config
+        prepared = self.prepare_config_write(config)
+        self.commit_config_write(prepared)
 
     def set_config_by_name(
         self,
@@ -120,8 +177,9 @@ def set_config_by_name(
         ):
             resolved_value = config_value[0] if config_value else None
 
-        setattr(self.config, config_name, resolved_value)
-        self.set_config(config=self.config)
+        candidate = self.config.model_copy(deep=True)
+        setattr(candidate, config_name, resolved_value)
+        self.set_config(config=candidate)
 
     # ==========================
     # Storage
diff --git a/reflexio/server/services/configurator/config_storage.py b/reflexio/server/services/configurator/config_storage.py
index 40a27cda8..56fcee509 100644
--- a/reflexio/server/services/configurator/config_storage.py
+++ b/reflexio/server/services/configurator/config_storage.py
@@ -1,9 +1,23 @@
 from abc import ABC, abstractmethod
+from collections.abc import Callable
+from dataclasses import dataclass
 from typing import Any
 
 from reflexio.models.config_schema import Config
 
 
+class ConfigWriteConflict(RuntimeError):  # noqa: N818 - domain name is part of contract
+    """Raised when another writer already owns an organization's config lock."""
+
+
+@dataclass(frozen=True)
+class ConfigPayloadUpdateResult:
+    """Result of one read-modify-write operation on a persisted config payload."""
+
+    payload: dict[str, Any]
+    changed: bool
+
+
 class ConfigStorage(ABC):
     """
     Abstract base class for configuration storage operations.
@@ -60,3 +74,22 @@ def get_version(self) -> tuple[str, Any] | None:
             TTL safety net.
         """
         return None
+
+    def update_config_payload(
+        self,
+        transform: Callable[[dict[str, Any]], dict[str, Any]],
+    ) -> ConfigPayloadUpdateResult:
+        """Apply ``transform`` to the latest payload and persist the result.
+
+        Backends with concurrency support override this method so the load,
+        transform, and save happen under one lock or transaction. The default
+        preserves compatibility for simple/test adapters.
+        """
+        current = self.load_config().model_dump(mode="json")
+        updated = transform(dict(current))
+        config = Config.model_validate(updated)
+        canonical = config.model_dump(mode="json")
+        if canonical == current:
+            return ConfigPayloadUpdateResult(payload=current, changed=False)
+        self.save_config(config)
+        return ConfigPayloadUpdateResult(payload=canonical, changed=True)
diff --git a/reflexio/server/services/configurator/local_file_config_storage.py b/reflexio/server/services/configurator/local_file_config_storage.py
index 507230ac9..ab9e48193 100644
--- a/reflexio/server/services/configurator/local_file_config_storage.py
+++ b/reflexio/server/services/configurator/local_file_config_storage.py
@@ -1,11 +1,13 @@
 import contextlib
 import copy
+import fcntl
 import json
 import logging
 import os
 import time
+from collections.abc import Callable
 from pathlib import Path
-from typing import Any
+from typing import Any, BinaryIO
 
 from reflexio.cli.paths import reflexio_home
 from reflexio.models.config_schema import (
@@ -15,7 +17,11 @@
     StorageConfigSQLite,
     validate_stored_config,
 )
-from reflexio.server.services.configurator.config_storage import ConfigStorage
+from reflexio.server.services.configurator.config_storage import (
+    ConfigPayloadUpdateResult,
+    ConfigStorage,
+    ConfigWriteConflict,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -74,7 +80,17 @@ def load_config(self) -> Config:
         """
         if not Path(self.config_file).exists():
             config = self.get_default_config()
-            self._save_config_to_local_dir(config=config)
+            try:
+                self.save_config(config=config)
+            except ConfigWriteConflict:
+                # A concurrent first load is already creating the file; a read
+                # must not fail on that, so serve defaults and let the other
+                # writer persist them.
+                logger.warning(
+                    "Skipped writing default config for org %s: another "
+                    "writer holds the config lock.",
+                    self.org_id,
+                )
             return config
 
         try:
@@ -105,9 +121,29 @@ def load_config(self) -> Config:
                     data["storage_config"] = self._default_storage_config().model_dump()
                 config = validate_stored_config(data)
                 if config.model_dump(mode="json") != original_payload:
+                    # Re-run the same normalization against the latest payload
+                    # under the write lock instead of persisting the pre-lock
+                    # snapshot, so a writer that completed after our read is
+                    # normalized rather than silently overwritten.
+                    def normalize_stored_payload(
+                        current: dict[str, Any],
+                    ) -> dict[str, Any]:
+                        normalized = dict(current)
+                        storage_cfg = normalized.get("storage_config")
+                        if (
+                            isinstance(storage_cfg, dict)
+                            and storage_cfg.get("type") == "disk"
+                        ):
+                            normalized["storage_config"] = (
+                                self._default_storage_config().model_dump()
+                            )
+                        return validate_stored_config(normalized).model_dump(
+                            mode="json"
+                        )
+
                     try:
-                        self._save_config_to_local_dir(config=config)
-                    except OSError:
+                        self.update_config_payload(normalize_stored_payload)
+                    except (OSError, ValueError, ConfigWriteConflict):
                         logger.exception(
                             "Loaded config from %s after normalizing its stored "
                             "schema, but could not rewrite the file; cleanup will "
@@ -131,12 +167,59 @@ def save_config(self, config: Config) -> None:
             config (Config): Configuration object to save
         """
         if self.base_dir and self.config_file:
-            self._save_config_to_local_dir(config=config)
+            self.update_config_payload(lambda _current: config.model_dump(mode="json"))
         else:
             print(
                 f"Cannot save config for org {self.org_id}: no local directory configured"
             )
 
+    def update_config_payload(
+        self,
+        transform: Callable[[dict[str, Any]], dict[str, Any]],
+    ) -> ConfigPayloadUpdateResult:
+        """Apply a payload update while holding a non-blocking file lock."""
+        if not (self.base_dir and self.config_file):
+            raise ValueError("base_dir and config_file must be set")
+
+        final_path = Path(self.config_file)
+        final_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_path = final_path.with_suffix(f"{final_path.suffix}.lock")
+        with lock_path.open("a+b") as lock_file:
+            self._acquire_lock(lock_file)
+            try:
+                current = self._read_payload_unlocked(final_path)
+                updated = transform(dict(current))
+                if updated == current:
+                    return ConfigPayloadUpdateResult(
+                        payload=current,
+                        changed=False,
+                    )
+                self._save_payload_to_local_dir(updated)
+                return ConfigPayloadUpdateResult(payload=updated, changed=True)
+            finally:
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+
+    @staticmethod
+    def _acquire_lock(lock_file: BinaryIO) -> None:
+        try:
+            fcntl.flock(
+                lock_file.fileno(),
+                fcntl.LOCK_EX | fcntl.LOCK_NB,
+            )
+        except BlockingIOError as exc:
+            raise ConfigWriteConflict(
+                "Configuration update already in progress"
+            ) from exc
+
+    @staticmethod
+    def _read_payload_unlocked(path: Path) -> dict[str, Any]:
+        if not path.exists():
+            return {}
+        data = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(data, dict):
+            raise ValueError("Configuration JSON must decode to an object")
+        return data
+
     def get_version(self) -> tuple[str, Any] | None:
         """Return the on-disk mtime of the org's config file, if it exists.
 
@@ -181,13 +264,20 @@ def _save_config_to_local_dir(self, config: Config) -> None:
         if not (self.base_dir and self.config_file):
             raise ValueError("base_dir and config_file must be set")
 
+        self._save_payload_to_local_dir(config.model_dump(mode="json"))
+
+    def _save_payload_to_local_dir(self, payload: dict[str, Any]) -> None:
+        """Atomically replace the local config file with a raw payload."""
         final_path = Path(self.config_file)
         final_path.parent.mkdir(parents=True, exist_ok=True)
         tmp_path = final_path.with_name(
             f"{final_path.name}.{os.getpid()}.{time.time_ns()}.tmp"
         )
         try:
-            tmp_path.write_text(config.model_dump_json(), encoding="utf-8")
+            tmp_path.write_text(
+                json.dumps(payload, separators=(",", ":")),
+                encoding="utf-8",
+            )
             tmp_path.replace(final_path)
         except OSError:
             with contextlib.suppress(OSError):
diff --git a/tests/lib/test_config_unit.py b/tests/lib/test_config_unit.py
index ca1a6089e..4cc10af43 100644
--- a/tests/lib/test_config_unit.py
+++ b/tests/lib/test_config_unit.py
@@ -7,10 +7,14 @@
 from typing import Any, cast
 from unittest.mock import MagicMock
 
+import pytest
+
 from reflexio.lib._config import ConfigMixin
 from reflexio.lib._dashboard import DashboardMixin
 from reflexio.models.api_schema.retriever_schema import GetDashboardStatsRequest
 from reflexio.models.config_schema import Config, StorageConfigSQLite
+from reflexio.server.services.configurator.base_configurator import PreparedConfigWrite
+from reflexio.server.services.configurator.config_storage import ConfigWriteConflict
 
 # ---------------------------------------------------------------------------
 # ConfigMixin helpers
@@ -26,6 +30,29 @@ def _make_config_mixin(*, storage_configured: bool = True) -> ConfigMixin:
     mock_request_context.org_id = "test_org"
     mock_request_context.storage = mock_storage if storage_configured else None
     mock_request_context.is_storage_configured.return_value = storage_configured
+    configurator = mock_request_context.configurator
+
+    def prepare(
+        value: Config | dict,
+        *,
+        partial: bool = False,
+    ) -> PreparedConfigWrite:
+        if isinstance(value, dict):
+            parsed = Config.model_validate(value)
+            payload = dict(value)
+        else:
+            parsed = value
+            payload = (
+                value.model_dump(mode="python") if isinstance(value, Config) else {}
+            )
+        return PreparedConfigWrite(
+            config=parsed,
+            payload=payload,
+            partial=partial,
+        )
+
+    configurator.prepare_config_write.side_effect = prepare
+    configurator.commit_config_write.return_value = True
 
     mixin.request_context = mock_request_context
     mixin.llm_client = MagicMock()
@@ -87,7 +114,7 @@ def test_set_config_success(self):
 
         assert response.success is True
         assert "successfully" in (response.msg or "").lower()
-        _get_configurator(mixin).set_config.assert_called_once()
+        _get_configurator(mixin).commit_config_write.assert_called_once()
 
     def test_set_config_storage_validation_fails(self):
         """Returns failure when storage validation fails."""
@@ -121,7 +148,8 @@ def test_set_config_skips_storage_init_when_storage_unchanged(self):
         assert response.success is True
         _get_configurator(mixin).is_storage_config_ready_to_test.assert_not_called()
         _get_configurator(mixin).test_and_init_storage_config.assert_not_called()
-        _get_configurator(mixin).set_config.assert_called_once_with(config)
+        prepared = _get_configurator(mixin).commit_config_write.call_args.args[0]
+        assert prepared.config is config
 
     def test_set_config_initializes_storage_when_storage_changed(self):
         """Changing storage still validates and initializes the new target."""
@@ -150,7 +178,8 @@ def test_set_config_initializes_storage_when_storage_changed(self):
         _get_configurator(mixin).test_and_init_storage_config.assert_called_once_with(
             storage_config=new_storage_config
         )
-        _get_configurator(mixin).set_config.assert_called_once_with(config)
+        prepared = _get_configurator(mixin).commit_config_write.call_args.args[0]
+        assert prepared.config is config
 
     def test_set_config_storage_not_ready(self):
         """Returns failure when storage config is incomplete."""
@@ -187,6 +216,25 @@ def test_set_config_preserves_existing_storage_config(self):
         # Verify storage_config was set to the existing one
         assert mock_config.storage_config == existing_storage_config
 
+    def test_partial_write_preserves_existing_storage_in_payload(self):
+        """The locked merge must not reintroduce an explicit redacted value."""
+        mixin = _make_config_mixin()
+        current_storage = StorageConfigSQLite(db_path="/var/data/current.db")
+        prepared = PreparedConfigWrite(
+            config=Config(storage_config=None, window_size=23),
+            payload={"storage_config": None, "window_size": 23},
+            partial=True,
+        )
+        _get_configurator(
+            mixin
+        ).get_current_storage_configuration.return_value = current_storage
+
+        response = mixin.set_config(prepared)
+
+        assert response.success is True
+        committed = _get_configurator(mixin).commit_config_write.call_args.args[0]
+        assert committed.payload["storage_config"] == current_storage
+
     def test_set_config_preserves_storage_on_managed_marker(self):
         """Round-tripping the redacted platform-managed marker that get_config()
         emits resolves to the real storage config instead of failing readiness.
@@ -223,11 +271,7 @@ def test_set_config_preserves_storage_on_managed_marker(self):
     def test_set_config_dict_input(self):
         """Accepts dict input and auto-converts to Config."""
         mixin = _make_config_mixin()
-        # normalize_config_payload is identity in the base configurator; the
-        # MagicMock default would otherwise return another MagicMock and break
-        # the **kwargs expansion below.
         payload = {"storage_config": {"db_path": "/var/data/test.db"}}
-        _get_configurator(mixin).normalize_config_payload.return_value = payload
         _get_configurator(
             mixin
         ).get_current_storage_configuration.return_value = MagicMock()
@@ -241,6 +285,20 @@ def test_set_config_dict_input(self):
 
         assert response.success is True
 
+    def test_set_config_propagates_write_conflict(self):
+        """Lock conflicts remain distinguishable for the HTTP route."""
+        mixin = _make_config_mixin()
+        config = Config(storage_config=StorageConfigSQLite())
+        _get_configurator(
+            mixin
+        ).get_current_storage_configuration.return_value = config.storage_config
+        _get_configurator(mixin).commit_config_write.side_effect = ConfigWriteConflict(
+            "busy"
+        )
+
+        with pytest.raises(ConfigWriteConflict, match="busy"):
+            mixin.set_config(config)
+
     def test_set_config_exception(self):
         """Returns failure on unexpected exception."""
         mixin = _make_config_mixin()
diff --git a/tests/server/api_endpoints/test_api_routes.py b/tests/server/api_endpoints/test_api_routes.py
index 119c36bb2..67ed3ba05 100644
--- a/tests/server/api_endpoints/test_api_routes.py
+++ b/tests/server/api_endpoints/test_api_routes.py
@@ -23,6 +23,33 @@
     UserProfile,
 )
 from reflexio.models.config_schema import Config, StorageConfigSQLite
+from reflexio.server.services.configurator.base_configurator import PreparedConfigWrite
+from reflexio.server.services.configurator.config_storage import ConfigWriteConflict
+
+
+def _prepare_config_write(
+    existing: Config,
+):
+    def prepare(
+        payload: Config | dict,
+        *,
+        partial: bool = False,
+    ) -> PreparedConfigWrite:
+        raw = (
+            payload.model_dump(mode="python")
+            if isinstance(payload, Config)
+            else dict(payload)
+        )
+        current = existing.model_dump(mode="python")
+        candidate = {**current, **raw} if partial else raw
+        config = Config.model_validate(candidate)
+        return PreparedConfigWrite(
+            config=config,
+            payload=raw,
+            partial=partial,
+        )
+
+    return prepare
 
 
 class TestHealthEndpoints:
@@ -269,6 +296,11 @@ class TestSetConfigRoute:
     def test_unknown_field_returns_422_before_set_config(
         self, client, patched_reflexio, mock_reflexio
     ):
+        existing = Config(storage_config=StorageConfigSQLite())
+        configurator = MagicMock()
+        configurator.prepare_config_write.side_effect = _prepare_config_write(existing)
+        mock_reflexio.request_context.configurator = configurator
+
         response = client.post(
             "/api/set_config",
             json={
@@ -305,6 +337,7 @@ def _existing_config() -> Config:
     def _wire_mock(self, mock_reflexio: MagicMock, existing: Config) -> None:
         configurator = MagicMock()
         configurator.get_config.return_value = existing
+        configurator.prepare_config_write.side_effect = _prepare_config_write(existing)
         mock_reflexio.request_context.configurator = configurator
         mock_reflexio.set_config.return_value = SetConfigResponse(
             success=True, msg="Configuration set successfully"
@@ -329,19 +362,23 @@ def test_partial_dict_succeeds(self, client, patched_reflexio, mock_reflexio):
         # The reflexio.set_config call receives a merged Config with the
         # new field flipped AND the existing storage_config preserved.
         assert mock_reflexio.set_config.call_count == 1
-        merged = mock_reflexio.set_config.call_args.args[0]
-        assert isinstance(merged, Config)
-        assert merged.window_size == 25
-        assert merged.storage_config == existing.storage_config
+        prepared = mock_reflexio.set_config.call_args.args[0]
+        assert isinstance(prepared, PreparedConfigWrite)
+        assert prepared.config.window_size == 25
+        assert prepared.config.storage_config == existing.storage_config
 
         # Cache invalidated on success.
         mock_invalidate.assert_called_once_with(org_id="test-org")
 
-    def test_no_op_patch_skips_set_config_and_cache_invalidation(
+    def test_no_op_patch_commits_under_lock_without_cache_invalidation(
         self, client, patched_reflexio, mock_reflexio
     ):
         existing = self._existing_config()
         self._wire_mock(mock_reflexio, existing)
+        mock_reflexio.set_config.return_value = SetConfigResponse(
+            success=True,
+            msg="Configuration unchanged",
+        )
 
         with patch(
             "reflexio.server.cache.reflexio_cache.invalidate_reflexio_cache"
@@ -356,7 +393,7 @@ def test_no_op_patch_skips_set_config_and_cache_invalidation(
             "success": True,
             "msg": "Configuration unchanged",
         }
-        mock_reflexio.set_config.assert_not_called()
+        mock_reflexio.set_config.assert_called_once()
         mock_invalidate.assert_not_called()
 
     def test_unknown_field_returns_422_before_set_config(
@@ -386,10 +423,10 @@ def test_replaces_nested_object_wholesale(
         )
 
         assert response.status_code == 200, response.text
-        merged = mock_reflexio.set_config.call_args.args[0]
-        assert isinstance(merged, Config)
-        assert isinstance(merged.storage_config, StorageConfigSQLite)
-        assert merged.storage_config.db_path == "/new/path.db"
+        prepared = mock_reflexio.set_config.call_args.args[0]
+        assert isinstance(prepared, PreparedConfigWrite)
+        assert isinstance(prepared.config.storage_config, StorageConfigSQLite)
+        assert prepared.config.storage_config.db_path == "/new/path.db"
 
     def test_does_not_invalidate_on_failure(
         self, client, patched_reflexio, mock_reflexio
@@ -398,6 +435,7 @@ def test_does_not_invalidate_on_failure(
         existing = self._existing_config()
         configurator = MagicMock()
         configurator.get_config.return_value = existing
+        configurator.prepare_config_write.side_effect = _prepare_config_write(existing)
         mock_reflexio.request_context.configurator = configurator
         mock_reflexio.set_config.return_value = SetConfigResponse(
             success=False, msg="storage validation failed"
@@ -481,12 +519,14 @@ def test_singular_extractor_configs_override_existing_config(
             )
 
         assert response.status_code == 200, response.text
-        merged = mock_reflexio.set_config.call_args.args[0]
-        assert isinstance(merged, Config)
-        assert merged.profile_extractor_config is not None
-        assert merged.profile_extractor_config.extractor_name == "profile"
-        assert merged.user_playbook_extractor_config is not None
-        assert merged.user_playbook_extractor_config.extractor_name == "playbook"
+        prepared = mock_reflexio.set_config.call_args.args[0]
+        assert isinstance(prepared, PreparedConfigWrite)
+        assert prepared.config.profile_extractor_config is not None
+        assert prepared.config.profile_extractor_config.extractor_name == "profile"
+        assert prepared.config.user_playbook_extractor_config is not None
+        assert (
+            prepared.config.user_playbook_extractor_config.extractor_name == "playbook"
+        )
 
     def test_null_extractor_configs_disable_existing_extractors(
         self, client, patched_reflexio, mock_reflexio
@@ -505,10 +545,10 @@ def test_null_extractor_configs_disable_existing_extractors(
             )
 
         assert response.status_code == 200, response.text
-        merged = mock_reflexio.set_config.call_args.args[0]
-        assert isinstance(merged, Config)
-        assert merged.profile_extractor_config is None
-        assert merged.user_playbook_extractor_config is None
+        prepared = mock_reflexio.set_config.call_args.args[0]
+        assert isinstance(prepared, PreparedConfigWrite)
+        assert prepared.config.profile_extractor_config is None
+        assert prepared.config.user_playbook_extractor_config is None
 
     def test_nested_config_preserved_when_patching_unrelated_field(
         self, client, patched_reflexio, mock_reflexio
@@ -524,12 +564,28 @@ def test_nested_config_preserved_when_patching_unrelated_field(
             )
 
         assert response.status_code == 200, response.text
-        merged = mock_reflexio.set_config.call_args.args[0]
-        assert isinstance(merged, Config)
+        prepared = mock_reflexio.set_config.call_args.args[0]
+        assert isinstance(prepared, PreparedConfigWrite)
         # The partial-touched field changed
-        assert merged.window_size == 25
-        assert merged.user_playbook_extractor_config is not None
-        agg = merged.user_playbook_extractor_config.aggregation_config
+        assert prepared.config.window_size == 25
+        assert prepared.config.user_playbook_extractor_config is not None
+        agg = prepared.config.user_playbook_extractor_config.aggregation_config
         assert agg is not None
         assert agg.min_cluster_size == 2
         assert agg.clustering_similarity == 0.45
+
+    def test_lock_conflict_returns_409_without_cache_invalidation(
+        self, client, patched_reflexio, mock_reflexio
+    ):
+        existing = self._existing_config()
+        self._wire_mock(mock_reflexio, existing)
+        mock_reflexio.set_config.side_effect = ConfigWriteConflict("busy")
+
+        with patch(
+            "reflexio.server.cache.reflexio_cache.invalidate_reflexio_cache"
+        ) as mock_invalidate:
+            response = client.post("/api/update_config", json={"window_size": 25})
+
+        assert response.status_code == 409
+        assert response.json()["detail"] == "Configuration update already in progress"
+        mock_invalidate.assert_not_called()
diff --git a/tests/server/services/configurator/test_config_storage_contract.py b/tests/server/services/configurator/test_config_storage_contract.py
index 43b38960f..a9078e310 100644
--- a/tests/server/services/configurator/test_config_storage_contract.py
+++ b/tests/server/services/configurator/test_config_storage_contract.py
@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import fcntl
 import json
 import tempfile
 from collections.abc import Generator
@@ -17,6 +18,8 @@
 import pytest
 
 from reflexio.models.config_schema import Config
+from reflexio.server.services.configurator.config_storage import ConfigWriteConflict
+from reflexio.server.services.configurator.configurator import DefaultConfigurator
 from reflexio.server.services.configurator.local_file_config_storage import (
     LocalFileConfigStorage,
 )
@@ -199,6 +202,104 @@ def boom(self: Path, *args, **kwargs) -> int:
         leaked = list(Path(tmp_path).rglob(f"{Path(storage.config_file).name}*.tmp"))
         assert not leaked, f"tmp file should be cleaned up on failure: {leaked}"
 
+    def test_save_config_fails_immediately_when_lock_is_held(self, tmp_path) -> None:
+        storage = LocalFileConfigStorage(org_id="locked-org", base_dir=str(tmp_path))
+        lock_path = Path(storage.config_file).with_suffix(".json.lock")
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with lock_path.open("a+b") as lock_file:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with pytest.raises(
+                ConfigWriteConflict,
+                match="already in progress",
+            ):
+                storage.save_config(storage.get_default_config())
+
+    def test_first_load_under_held_lock_returns_default_config(self, tmp_path) -> None:
+        """A concurrent first-load must serve defaults, not raise a conflict."""
+        storage = LocalFileConfigStorage(org_id="fresh-org", base_dir=str(tmp_path))
+        lock_path = Path(storage.config_file).with_suffix(".json.lock")
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with lock_path.open("a+b") as lock_file:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            loaded = storage.load_config()
+
+        assert loaded == storage.get_default_config()
+
+    def test_schema_cleanup_conflict_returns_valid_stored_config(
+        self, tmp_path
+    ) -> None:
+        storage = LocalFileConfigStorage(org_id="read-locked", base_dir=str(tmp_path))
+        payload = storage.get_default_config().model_dump(mode="json")
+        payload["retired_field"] = "ignored"
+        payload["agent_context_prompt"] = "user value survives"
+        config_path = Path(storage.config_file)
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        config_path.write_text(json.dumps(payload), encoding="utf-8")
+        lock_path = config_path.with_suffix(".json.lock")
+
+        with lock_path.open("a+b") as lock_file:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+            loaded = storage.load_config()
+
+        assert loaded.storage_config == storage.get_default_config().storage_config
+        assert loaded.agent_context_prompt == "user value survives"
+        assert json.loads(config_path.read_text())["retired_field"] == "ignored"
+
+    def test_schema_cleanup_preserves_concurrent_newer_write(
+        self, tmp_path, monkeypatch
+    ) -> None:
+        """The cleanup rewrite must normalize the latest payload, not the
+        pre-lock snapshot, so a write completing between the read and the
+        lock is cleaned rather than silently overwritten."""
+        storage = LocalFileConfigStorage(org_id="cleanup-race", base_dir=str(tmp_path))
+        stale = storage.get_default_config().model_dump(mode="json")
+        stale["retired_field"] = "drop me"
+        config_path = Path(storage.config_file)
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        config_path.write_text(json.dumps(stale), encoding="utf-8")
+
+        newer = dict(stale)
+        newer["agent_context_prompt"] = "written after the pre-lock read"
+        monkeypatch.setattr(
+            storage,
+            "_read_payload_unlocked",
+            lambda _path: dict(newer),
+        )
+
+        loaded = storage.load_config()
+
+        persisted = json.loads(config_path.read_text(encoding="utf-8"))
+        assert persisted["agent_context_prompt"] == "written after the pre-lock read"
+        assert "retired_field" not in persisted
+        # The returned config still reflects the payload this reader saw.
+        assert loaded.agent_context_prompt is None
+
+    def test_sequential_partial_commits_reload_latest_payload(self, tmp_path) -> None:
+        configurator = DefaultConfigurator(
+            org_id="partial-org",
+            base_dir=str(tmp_path),
+        )
+        first = configurator.prepare_config_write(
+            {"agent_context_prompt": "first"},
+            partial=True,
+        )
+        second = configurator.prepare_config_write(
+            {"window_size": 23},
+            partial=True,
+        )
+
+        assert configurator.commit_config_write(first) is True
+        assert configurator.commit_config_write(second) is True
+
+        reloaded = DefaultConfigurator(
+            org_id="partial-org",
+            base_dir=str(tmp_path),
+        ).get_config()
+        assert reloaded.agent_context_prompt == "first"
+        assert reloaded.window_size == 23
+
     def test_get_version_changes_after_save(
         self, config_storage: ConfigStorage
     ) -> None: