feat(BA-6554): add app_config_fragment service layer (#12358)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jopemachine

changes/12358.feature.md

@@ -0,0 +1 @@
+Add the app_config_fragment service layer (admin create / update / purge with the allow-list write-gate).

docs/manager/graphql-reference/supergraph.graphql

@@ -15256,6 +15256,7 @@ enum RBACElementType
   APP_CONFIG @join__enumValue(graph: STRAWBERRY)
   APP_CONFIG_DEFINITION @join__enumValue(graph: STRAWBERRY)
   APP_CONFIG_ALLOW_LIST @join__enumValue(graph: STRAWBERRY)
+  APP_CONFIG_FRAGMENT @join__enumValue(graph: STRAWBERRY)
   MODEL_CARD @join__enumValue(graph: STRAWBERRY)
   RESOURCE_PRESET @join__enumValue(graph: STRAWBERRY)
   USER_RESOURCE_POLICY @join__enumValue(graph: STRAWBERRY)

docs/manager/graphql-reference/v2-schema.graphql

@@ -10258,6 +10258,7 @@ enum RBACElementType {
   APP_CONFIG
   APP_CONFIG_DEFINITION
   APP_CONFIG_ALLOW_LIST
+  APP_CONFIG_FRAGMENT
   MODEL_CARD
   RESOURCE_PRESET
   USER_RESOURCE_POLICY

src/ai/backend/common/data/permission/types.py

@@ -104,6 +104,7 @@ class EntityType(enum.StrEnum):
     AGENT = "agent"
     APP_CONFIG_DEFINITION = "app_config_definition"
     APP_CONFIG_ALLOW_LIST = "app_config_allow_list"
+    APP_CONFIG_FRAGMENT = "app_config_fragment"
     AUTH = "auth"
     PROJECT_ADMIN_PAGE = "project_admin_page"
     DOMAIN_ADMIN_PAGE = "domain_admin_page"
@@ -408,6 +409,7 @@ class RBACElementType(enum.StrEnum):
     APP_CONFIG = "app_config"
     APP_CONFIG_DEFINITION = "app_config_definition"
     APP_CONFIG_ALLOW_LIST = "app_config_allow_list"
+    APP_CONFIG_FRAGMENT = "app_config_fragment"
     MODEL_CARD = "model_card"
 
     # === Root-query-enabled entities (superadmin-only) ===

src/ai/backend/common/dto/manager/v2/rbac/types.py

@@ -99,6 +99,7 @@ class RBACElementTypeDTO(StrEnum):
     APP_CONFIG = "app_config"
     APP_CONFIG_DEFINITION = "app_config_definition"
     APP_CONFIG_ALLOW_LIST = "app_config_allow_list"
+    APP_CONFIG_FRAGMENT = "app_config_fragment"
     MODEL_CARD = "model_card"
 
     # Root-query-enabled entities (superadmin-only)

src/ai/backend/manager/api/gql/rbac/types/permission.py

@@ -232,6 +232,7 @@ async def scope(
                 | RBACElementType.APP_CONFIG
                 | RBACElementType.APP_CONFIG_DEFINITION
                 | RBACElementType.APP_CONFIG_ALLOW_LIST
+                | RBACElementType.APP_CONFIG_FRAGMENT
                 | RBACElementType.RESOURCE_PRESET
                 | RBACElementType.USER_RESOURCE_POLICY
                 | RBACElementType.KEYPAIR_RESOURCE_POLICY

src/ai/backend/manager/errors/app_config.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
-from ai.backend.manager.errors.common import ObjectNotFound
+from ai.backend.manager.errors.common import GenericForbidden, ObjectNotFound
 
 __all__ = (
     "AppConfigAllowListNotFound",
     "AppConfigDefinitionNotFound",
     "AppConfigFragmentNotFound",
+    "AppConfigFragmentWriteNotAllowed",
 )
 
 
@@ -22,3 +23,14 @@ class AppConfigAllowListNotFound(ObjectNotFound):
 class AppConfigFragmentNotFound(ObjectNotFound):
     error_type = "https://api.backend.ai/probs/app-config-fragment-not-found"
     object_name = "app config fragment"
+
+
+class AppConfigFragmentWriteNotAllowed(GenericForbidden):
+    """A fragment write was rejected by the write-gate.
+
+    Raised when the target ``config_name`` is not registered, or no app_config_allow_list
+    row exists for the target ``(config_name, scope_type)`` pair.
+    """
+
+    error_type = "https://api.backend.ai/probs/app-config-fragment-write-not-allowed"
+    error_title = "App config fragment write is not allowed for this config/scope."

src/ai/backend/manager/repositories/app_config_fragment/db_source/db_source.py

@@ -16,14 +16,24 @@
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
-from ai.backend.manager.errors.app_config import AppConfigFragmentNotFound
+from ai.backend.manager.errors.app_config import (
+    AppConfigFragmentNotFound,
+    AppConfigFragmentWriteNotAllowed,
+)
+from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
 from ai.backend.manager.models.app_config_definition.row import AppConfigDefinitionRow
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.app_config_fragment.creators import (
     AppConfigFragmentCreatorSpec,
 )
-from ai.backend.manager.repositories.base import BatchQuerier, Purger, Querier, Updater
+from ai.backend.manager.repositories.base import (
+    BatchQuerier,
+    ExistsQuerier,
+    Purger,
+    Querier,
+    Updater,
+)
 from ai.backend.manager.repositories.base.creator import NextValuePolicy
 from ai.backend.manager.repositories.ops import DBOpsProvider
 
@@ -61,7 +71,11 @@ def __init__(self, ops_provider: DBOpsProvider) -> None:
         self._ops = ops_provider
 
     @app_config_fragment_db_source_resilience.apply()
-    async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentData:
+    async def create(
+        self,
+        spec: AppConfigFragmentCreatorSpec,
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
         policy = NextValuePolicy(
             column=AppConfigFragmentRow.rank,
             scope_condition=lambda: AppConfigFragmentRow.config_name == spec.config_name,
@@ -70,7 +84,14 @@ async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentD
             ),
             gap=RANK_GAP,
         )
+        # ``only_if`` (built by the caller) and the write run in one transaction, so the gate
+        # check and the write commit atomically — no check-then-write race.
         async with self._ops.write_ops() as w:
+            if not await w.exists(only_if):
+                raise AppConfigFragmentWriteNotAllowed(
+                    f"Writing app config {spec.config_name!r} at scope "
+                    f"{spec.scope_type.value!r} is not allowed."
+                )
             created = await w.create_with_next_value(policy, spec)
             return created.row.to_data()
 
@@ -83,16 +104,36 @@ async def get_by_id(self, fragment_id: AppConfigFragmentID) -> AppConfigFragment
             return result.row.to_data()
 
     @app_config_fragment_db_source_resilience.apply()
-    async def update(self, updater: Updater[AppConfigFragmentRow]) -> AppConfigFragmentData:
+    async def update(
+        self,
+        updater: Updater[AppConfigFragmentRow],
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        # Gate first, then write — both in one transaction so the check and the write commit
+        # atomically. A missing fragment surfaces as the update returning None below.
         async with self._ops.write_ops() as w:
+            if not await w.exists(only_if):
+                raise AppConfigFragmentWriteNotAllowed(
+                    f"Writing app config fragment {updater.pk_value} is not allowed."
+                )
             result = await w.update(updater)
             if result is None:
                 raise AppConfigFragmentNotFound(f"App config fragment {updater.pk_value} not found")
             return result.row.to_data()
 
     @app_config_fragment_db_source_resilience.apply()
-    async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragmentData:
+    async def purge(
+        self,
+        purger: Purger[AppConfigFragmentRow],
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        # Gate first, then write — both in one transaction so the check and the write commit
+        # atomically. A missing fragment surfaces as the purge returning None below.
         async with self._ops.write_ops() as w:
+            if not await w.exists(only_if):
+                raise AppConfigFragmentWriteNotAllowed(
+                    f"Writing app config fragment {purger.pk_value} is not allowed."
+                )
             result = await w.purge(purger)
             if result is None:
                 raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")

src/ai/backend/manager/repositories/app_config_fragment/repository.py

@@ -12,6 +12,7 @@
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
+from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.app_config_fragment.creators import (
@@ -20,7 +21,7 @@
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
-from ai.backend.manager.repositories.base import BatchQuerier, Purger, Updater
+from ai.backend.manager.repositories.base import BatchQuerier, ExistsQuerier, Purger, Updater
 from ai.backend.manager.repositories.ops import DBOpsProvider
 
 __all__ = ("AppConfigFragmentRepository",)
@@ -54,20 +55,32 @@ def __init__(self, ops_provider: DBOpsProvider) -> None:
         self._db_source = AppConfigFragmentDBSource(ops_provider)
 
     @app_config_fragment_repository_resilience.apply()
-    async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentData:
-        return await self._db_source.create(spec)
+    async def create(
+        self,
+        spec: AppConfigFragmentCreatorSpec,
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        return await self._db_source.create(spec, only_if)
 
     @app_config_fragment_repository_resilience.apply()
     async def get_by_id(self, fragment_id: AppConfigFragmentID) -> AppConfigFragmentData:
         return await self._db_source.get_by_id(fragment_id)
 
     @app_config_fragment_repository_resilience.apply()
-    async def update(self, updater: Updater[AppConfigFragmentRow]) -> AppConfigFragmentData:
-        return await self._db_source.update(updater)
+    async def update(
+        self,
+        updater: Updater[AppConfigFragmentRow],
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        return await self._db_source.update(updater, only_if)
 
     @app_config_fragment_repository_resilience.apply()
-    async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragmentData:
-        return await self._db_source.purge(purger)
+    async def purge(
+        self,
+        purger: Purger[AppConfigFragmentRow],
+        only_if: ExistsQuerier[AppConfigAllowListRow],
+    ) -> AppConfigFragmentData:
+        return await self._db_source.purge(purger, only_if)
 
     @app_config_fragment_repository_resilience.apply()
     async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchResult:

src/ai/backend/manager/repositories/app_config_fragment/types.py

@@ -8,44 +8,67 @@
 
 import sqlalchemy as sa
 
-from ai.backend.manager.errors.app_config import AppConfigDefinitionNotFound
-from ai.backend.manager.models.app_config_definition.row import AppConfigDefinitionRow
+from ai.backend.common.data.app_config.types import AppConfigScopeType
+from ai.backend.common.identifier.domain import DomainID
+from ai.backend.common.identifier.user import UserID
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.clauses import QueryCondition
 from ai.backend.manager.models.scopes import ExistenceCheck, SearchScope
 
-__all__ = ("ConfigNameSearchScope",)
+__all__ = (
+    "DomainAppConfigFragmentSearchScope",
+    "UserAppConfigFragmentSearchScope",
+)
 
 
 @dataclass(frozen=True)
-class ConfigNameSearchScope(SearchScope):
-    """Fragments belonging to a single ``config_name``.
+class DomainAppConfigFragmentSearchScope(SearchScope):
+    """Fragments written at one domain scope (``scope_type == domain``, ``scope_id == domain_id``).
 
-    One scope = all fragments registered under one app config name, regardless of their
-    ``scope_type`` (public / domain / user). The existence check rejects an unregistered
-    ``config_name`` up front so a scoped search cannot silently return nothing.
+    One scope = one item of a scoped fragment query; the repository combines multiple
+    scopes with ``OR``. ``existence_checks`` is empty by ``SearchableActionTarget``
+    convention — RBAC already gates scope reachability.
     """
 
-    config_name: str
+    domain_id: DomainID
 
     @override
     def to_condition(self) -> QueryCondition:
-        config_name = self.config_name
+        domain_id = self.domain_id
 
         def inner() -> sa.sql.expression.ColumnElement[bool]:
-            return AppConfigFragmentRow.config_name == config_name
+            return sa.and_(
+                AppConfigFragmentRow.scope_type == AppConfigScopeType.DOMAIN,
+                AppConfigFragmentRow.scope_id == str(domain_id),
+            )
 
         return inner
 
     @property
     @override
     def existence_checks(self) -> Sequence[ExistenceCheck[Any]]:
-        return (
-            ExistenceCheck(
-                column=AppConfigDefinitionRow.config_name,
-                value=self.config_name,
-                error=AppConfigDefinitionNotFound(
-                    f"App config definition {self.config_name!r} not found"
-                ),
-            ),
-        )
+        return ()
+
+
+@dataclass(frozen=True)
+class UserAppConfigFragmentSearchScope(SearchScope):
+    """Fragments written at one user scope (``scope_type == user``, ``scope_id == user_id``)."""
+
+    user_id: UserID
+
+    @override
+    def to_condition(self) -> QueryCondition:
+        user_id = self.user_id
+
+        def inner() -> sa.sql.expression.ColumnElement[bool]:
+            return sa.and_(
+                AppConfigFragmentRow.scope_type == AppConfigScopeType.USER,
+                AppConfigFragmentRow.scope_id == str(user_id),
+            )
+
+        return inner
+
+    @property
+    @override
+    def existence_checks(self) -> Sequence[ExistenceCheck[Any]]:
+        return ()