fix(registry): preset/model-endpoint mutation integrity (rename, default, delete)

openrag/core/ports/model_endpoint_repo.py

@@ -30,3 +30,10 @@ async def delete(self, name: str, model_type: str) -> bool: ...
 
     @abstractmethod
     async def set_default(self, model_type: str, name: str) -> None: ...
+
+    @abstractmethod
+    async def delete_and_promote_default(self, name: str, model_type: str) -> tuple[str, str | None]:
+        """Atomically delete an endpoint and, if it was the default, promote a
+        survivor. Decides under a row lock. Returns ``(status, promoted_name)``
+        where status is ``"not_found" | "last" | "ok"``."""
+        ...

openrag/services/orchestrators/model_endpoint_service.py

@@ -246,6 +246,11 @@ async def update_model_endpoint(self, name: str, model_type: str, **fields: obje
             self._invalidate_client_cache(model_type, name)
 
         self._invalidate_client_cache(model_type, effective_name)
+        if existing.is_default:
+            # The 'default' alias resolves to this endpoint, so a client built under
+            # the 'default' cache key is now stale too. Evict it alongside the named
+            # entry (set_default already does this; update must match).
+            self._invalidate_client_cache(model_type, "default")
         await self.load_all()
         return await self._repo.get(effective_name, model_type) or (updated or existing)
 
@@ -255,16 +260,21 @@ async def delete_model_endpoint(self, name: str, model_type: str) -> None:
         Raises 404 if not found, 422 if it is the last endpoint of its type
         (would leave components with no fallback).
         """
-        existing = await self._repo.get(name, model_type)
-        if existing is None:
+        # The last-endpoint guard and the survivor/default choice are made INSIDE
+        # the repo's locked transaction (not from a stale snapshot here), so
+        # concurrent deletes of the same type can't both pass the count check or
+        # promote an already-deleted survivor — which would leave the type with no
+        # endpoint / no default. The repo reports what happened.
+        status, promoted = await self._repo.delete_and_promote_default(name, model_type)
+        if status == "not_found":
             raise NotFoundError(f"Endpoint '{name}' of type '{model_type}' not found.")
-
-        all_of_type = await self._repo.list_all(model_type=model_type)
-        if len(all_of_type) <= 1:
+        if status == "last":
             raise ValidationError(f"Cannot delete the last '{model_type}' endpoint. Register a replacement first.")
 
-        await self._repo.delete(name, model_type)
         self._invalidate_client_cache(model_type, name)
+        if promoted is not None:
+            # The deleted endpoint was the default; its 'default' alias client is now stale.
+            self._invalidate_client_cache(model_type, "default")
         await self.load_all()
 
     async def set_default(self, model_type: str, name: str) -> None:

openrag/services/orchestrators/preset_service.py

@@ -27,6 +27,12 @@
 
 _VALID_PRESET_TYPES = frozenset({"indexation", "retrieval"})
 
+# The 'default' preset is the system fallback: partitions.{indexation,retrieval}_preset
+# default to this name (schema server_default + create_partition), and
+# resolve_partition_row raises if it is missing. It must always exist under this
+# name, so renaming it away or deleting it is rejected. Editing its config is fine.
+_RESERVED_PRESET_NAME = "default"
+
 _DEFAULT_SEEDS: dict[str, dict[str, dict[str, Any]]] = {
     # ``enable_contextualization`` is intentionally omitted from the ``default``
     # indexation preset below — it is injected at seed time from the global
@@ -210,6 +216,15 @@ async def update_preset(self, name: str, preset_type: str, **fields: object) ->
         effective_name = name
 
         if new_name and new_name != name:
+            if name == _RESERVED_PRESET_NAME:
+                raise ValidationError(
+                    f"Cannot rename the reserved '{_RESERVED_PRESET_NAME}' {preset_type} preset: "
+                    "it is the system fallback every partition resolves to."
+                )
+            if await self._repo.get(new_name, preset_type) is not None:
+                raise ValidationError(
+                    f"Cannot rename preset to '{new_name}': a {preset_type} preset with that name already exists."
+                )
             effective_name = new_name
             result = await self._repo.rename(name, effective_name, preset_type, effective_config)
         else:
@@ -230,6 +245,12 @@ async def delete_preset(self, name: str, preset_type: str) -> None:
         if existing is None:
             raise NotFoundError(f"Preset '{name}' of type '{preset_type}' not found.")
 
+        if name == _RESERVED_PRESET_NAME:
+            raise ValidationError(
+                f"Cannot delete the reserved '{_RESERVED_PRESET_NAME}' {preset_type} preset: "
+                "it is the system fallback every partition resolves to."
+            )
+
         count = await self._repo.count_partitions_using(name, preset_type)
         if count > 0:
             raise ValidationError(

openrag/services/persistence/model_endpoint_repo.py

@@ -132,5 +132,46 @@ async def set_default(self, model_type: str, name: str) -> None:
                     model_type,
                 )
 
+    async def delete_and_promote_default(self, name: str, model_type: str) -> tuple[str, str | None]:
+        # Lock every row of this model_type (FOR UPDATE) so concurrent deletes of
+        # the same type serialize, then make the last-endpoint guard and survivor
+        # choice from the LOCKED, current state — not a stale snapshot. Otherwise two
+        # concurrent deletes could each pass a snapshot count check and remove the
+        # last row, or promote a survivor that another tx just deleted, leaving the
+        # type with no endpoint / no default. Returns (status, promoted_name):
+        # status is "not_found" | "last" | "ok"; promoted_name is set only when the
+        # deleted row was the default and a survivor was promoted.
+        async with self.pool.acquire() as conn:
+            async with conn.transaction():
+                rows = await conn.fetch(
+                    "SELECT name, is_default FROM model_endpoints WHERE model_type = $1 ORDER BY name FOR UPDATE",
+                    model_type,
+                )
+                names = [r["name"] for r in rows]
+                if name not in names:
+                    return ("not_found", None)
+                if len(names) <= 1:
+                    return ("last", None)
+                was_default = next(r["is_default"] for r in rows if r["name"] == name)
+                await conn.execute(
+                    "DELETE FROM model_endpoints WHERE name = $1 AND model_type = $2",
+                    name,
+                    model_type,
+                )
+                promoted: str | None = None
+                if was_default:
+                    promoted = next(n for n in names if n != name)  # deterministic: first by name
+                    await conn.execute(
+                        "UPDATE model_endpoints SET is_default = false, updated_at = now() WHERE model_type = $1",
+                        model_type,
+                    )
+                    await conn.execute(
+                        "UPDATE model_endpoints SET is_default = true, updated_at = now() "
+                        "WHERE name = $1 AND model_type = $2",
+                        promoted,
+                        model_type,
+                    )
+                return ("ok", promoted)
+
 
 __all__ = ["PgModelEndpointRepository"]

openrag/services/persistence/preset_repo.py

@@ -11,6 +11,7 @@
 from typing import TYPE_CHECKING
 
 from core.ports.preset_repo import PresetRepository
+from core.utils.exceptions import NotFoundError
 
 if TYPE_CHECKING:
     import asyncpg
@@ -79,18 +80,32 @@ async def upsert(self, name: str, preset_type: str, config: dict) -> dict:
         return self._row_to_dict(rec)
 
     async def rename(self, old_name: str, new_name: str, preset_type: str, config: dict) -> dict:
+        # Rename in place with a plain UPDATE (not an upsert): the
+        # (name, preset_type) unique constraint then rejects a collision with an
+        # existing target name instead of an upsert silently overwriting it, and
+        # the row keeps its identity/created_at. Partitions reference a preset by
+        # its name string (partitions.{col}) with no FK, so referencing partitions
+        # are repointed in the same transaction or they would orphan.
+        col = _preset_column(preset_type)
         async with self.pool.acquire() as conn:
             async with conn.transaction():
-                await conn.execute(
-                    "DELETE FROM pipeline_presets WHERE name = $1 AND preset_type = $2",
+                rec = await conn.fetchrow(
+                    "UPDATE pipeline_presets SET name = $2, config = $3::jsonb, updated_at = now() "
+                    "WHERE name = $1 AND preset_type = $4 RETURNING *",
                     old_name,
+                    new_name,
+                    config,
                     preset_type,
                 )
-                rec = await conn.fetchrow(
-                    _UPSERT_PRESET_SQL,
+                if rec is None:
+                    # old_name vanished between the service's existence check and
+                    # this UPDATE (concurrent delete) — fail cleanly instead of
+                    # blowing up in _row_to_dict(None).
+                    raise NotFoundError(f"Preset '{old_name}' of type '{preset_type}' not found.")
+                await conn.execute(
+                    f"UPDATE partitions SET {col} = $1 WHERE {col} = $2",
                     new_name,
-                    preset_type,
-                    config,
+                    old_name,
                 )
         return self._row_to_dict(rec)
 

tests/unit/services/orchestrators/test_model_endpoint_service.py

@@ -84,6 +84,23 @@ async def set_default(self, model_type: str, name: str) -> None:
         for key, row in list(self._store.items()):
             self._store[key] = row.model_copy(update={"is_default": key[0] == name and key[1] == model_type})
 
+    async def delete_and_promote_default(self, name: str, model_type: str) -> tuple[str, str | None]:
+        names = sorted(k[0] for k in self._store if k[1] == model_type)
+        self.calls.append(("delete_and_promote_default", (name, model_type)))
+        if name not in names:
+            return ("not_found", None)
+        if len(names) <= 1:
+            return ("last", None)
+        was_default = self._store[(name, model_type)].is_default
+        self._store.pop((name, model_type), None)
+        promoted = None
+        if was_default:
+            promoted = next(n for n in names if n != name)
+            for key, row in list(self._store.items()):
+                if key[1] == model_type:
+                    self._store[key] = row.model_copy(update={"is_default": key[0] == promoted})
+        return ("ok", promoted)
+
 
 def _make_service(repo=None, rows=None, settings=None):
     from core.config.root import Settings
@@ -387,6 +404,38 @@ async def test_update_model_endpoint_renames_and_evicts_cache():
     assert ("new-name", "embedder") in repo._store
 
 
+@pytest.mark.asyncio
+async def test_update_default_endpoint_evicts_default_alias_cache():
+    # Updating the current default must evict the 'default' cache key too, not just
+    # the named one — else callers that resolved 'default' keep a stale client.
+    existing = _make_row(name="jina", is_default=True)
+    repo = _FakeEndpointRepo(rows=[existing])
+    cache: dict = {"jina": object(), "default": object()}
+    svc = _make_service(repo)
+    svc._client_caches["embedder"] = cache
+
+    await svc.update_model_endpoint("jina", "embedder", endpoint="http://new:8000/v1")
+
+    assert "jina" not in cache
+    assert "default" not in cache
+
+
+@pytest.mark.asyncio
+async def test_update_non_default_endpoint_keeps_default_alias_cache():
+    # A non-default endpoint update must NOT evict the unrelated 'default' client.
+    rows = [_make_row(name="e5", is_default=False), _make_row(name="jina", is_default=True)]
+    repo = _FakeEndpointRepo(rows=rows)
+    sentinel = object()
+    cache: dict = {"e5": object(), "default": sentinel}
+    svc = _make_service(repo)
+    svc._client_caches["embedder"] = cache
+
+    await svc.update_model_endpoint("e5", "embedder", endpoint="http://new:8000/v1")
+
+    assert "e5" not in cache
+    assert cache.get("default") is sentinel
+
+
 # ------------------------------------------------------------------
 # delete_model_endpoint
 # ------------------------------------------------------------------
@@ -428,6 +477,43 @@ async def test_delete_model_endpoint_removes_row_and_evicts_cache():
     assert "jina" not in cache
 
 
+@pytest.mark.asyncio
+async def test_delete_default_endpoint_promotes_survivor_and_evicts_default_cache():
+    # Deleting the default must promote a surviving endpoint so the 'default' alias
+    # keeps resolving (load_all only adds it for an is_default row), and evict the
+    # now-stale 'default' client.
+    rows = [
+        _make_row(name="jina", is_default=True),
+        _make_row(name="e5", is_default=False),
+    ]
+    repo = _FakeEndpointRepo(rows=rows)
+    cache: dict = {"jina": object(), "default": object()}
+    svc = _make_service(repo)
+    svc._client_caches["embedder"] = cache
+
+    await svc.delete_model_endpoint("jina", "embedder")
+
+    assert ("jina", "embedder") not in repo._store
+    assert "default" not in cache
+    # e5 is the lone survivor -> promoted so 'default' still resolves.
+    assert repo._store[("e5", "embedder")].is_default is True
+
+
+@pytest.mark.asyncio
+async def test_delete_non_default_endpoint_leaves_default_untouched():
+    rows = [
+        _make_row(name="jina", is_default=True),
+        _make_row(name="e5", is_default=False),
+    ]
+    repo = _FakeEndpointRepo(rows=rows)
+    svc = _make_service(repo)
+
+    await svc.delete_model_endpoint("e5", "embedder")
+
+    # Deleting a non-default endpoint leaves the existing default in place.
+    assert repo._store[("jina", "embedder")].is_default is True
+
+
 # ------------------------------------------------------------------
 # set_default
 # ------------------------------------------------------------------

tests/unit/services/orchestrators/test_preset_service.py

@@ -334,6 +334,26 @@ async def test_update_preset_renames_preset():
     assert not any(call[0] == "delete" for call in repo.calls)
 
 
+@pytest.mark.asyncio
+async def test_update_preset_rename_rejects_existing_target_name():
+    from core.utils.exceptions import ValidationError
+
+    rows = [
+        _make_row("old-name", "retrieval", _VALID_RET_CONFIG),
+        _make_row("taken", "retrieval", _VALID_RET_CONFIG),
+    ]
+    repo = _FakePresetRepo(rows=rows)
+    svc = _make_service(repo)
+
+    with pytest.raises(ValidationError, match="already exists"):
+        await svc.update_preset("old-name", "retrieval", new_name="taken")
+
+    # The rename must not run — the existing 'taken' preset is left intact.
+    assert not any(call[0] == "rename" for call in repo.calls)
+    assert ("old-name", "retrieval") in repo._store
+    assert ("taken", "retrieval") in repo._store
+
+
 @pytest.mark.asyncio
 async def test_update_preset_calls_partition_service():
     existing = _make_row("default", "retrieval", _VALID_RET_CONFIG)
@@ -376,13 +396,55 @@ async def test_delete_preset_raises_404_when_missing():
 async def test_delete_preset_raises_422_when_in_use():
     from core.utils.exceptions import ValidationError
 
-    existing = _make_row("default", "indexation")
+    existing = _make_row("legal", "indexation")
     repo = _FakePresetRepo(rows=[existing])
-    repo._partition_counts[("default", "indexation")] = 2
+    repo._partition_counts[("legal", "indexation")] = 2
     svc = _make_service(repo)
 
     with pytest.raises(ValidationError, match="2 partition"):
+        await svc.delete_preset("legal", "indexation")
+
+
+@pytest.mark.asyncio
+async def test_delete_preset_rejects_reserved_default():
+    from core.utils.exceptions import ValidationError
+
+    existing = _make_row("default", "indexation")
+    repo = _FakePresetRepo(rows=[existing])
+    svc = _make_service(repo)
+
+    # The reserved 'default' preset is the partition fallback — deletion is blocked
+    # even when no partition currently references it, before the in-use count check.
+    with pytest.raises(ValidationError, match="reserved"):
         await svc.delete_preset("default", "indexation")
+    assert ("default", "indexation") in repo._store
+
+
+@pytest.mark.asyncio
+async def test_update_preset_rejects_renaming_reserved_default():
+    from core.utils.exceptions import ValidationError
+
+    existing = _make_row("default", "retrieval", _VALID_RET_CONFIG)
+    repo = _FakePresetRepo(rows=[existing])
+    svc = _make_service(repo)
+
+    with pytest.raises(ValidationError, match="reserved"):
+        await svc.update_preset("default", "retrieval", new_name="custom")
+    assert not any(call[0] == "rename" for call in repo.calls)
+    assert ("default", "retrieval") in repo._store
+
+
+@pytest.mark.asyncio
+async def test_update_preset_allows_editing_reserved_default_config():
+    # Editing the default preset's config (no rename) must stay allowed.
+    existing = _make_row("default", "indexation", _VALID_IDX_CONFIG)
+    repo = _FakePresetRepo(rows=[existing])
+    svc = _make_service(repo)
+
+    edited = {**_VALID_IDX_CONFIG, "enable_topic_tagging": False}
+    await svc.update_preset("default", "indexation", config=edited)
+
+    assert repo._store[("default", "indexation")]["config"]["enable_topic_tagging"] is False
 
 
 @pytest.mark.asyncio