fix: make configuration writes strict and atomic

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)}"

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

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`

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,21 +99,21 @@ 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,
             detail=exc.errors(),
         ) 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,17 +162,15 @@ 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
     # would replace a nested extractor object with an incomplete dict (e.g.
     # {"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
 

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

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)