feat(BA-6626): app_config_fragment bulk repository layer

Build the app_config_fragment bulk repository on the conditional-bulk primitives (#12429):

- AppConfigFragmentCreatorSpec as a plain CreatorSpec with scope-based rank
  (public < domain < user).
- bulk_create / bulk_update / bulk_purge in db_source + repository, returning the
  AppConfigFragmentBulkWriteResult data type (succeeded + failed[index, message]) — partial
  success via the WriteOps.bulk_conditional_*_partial primitives.
- Single create rewritten to gate + plain create (no next-value); GatedWrite removed.
- Repository unit tests.

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

Author: jopemachine

changes/12426.misc.md

@@ -0,0 +1 @@
+Add `app_config_fragment` bulk repository operations (`bulk_create`/`bulk_update`/`bulk_purge` with per-item partial success): bulk create pairs each item with its allow-list write-gate, bulk update/purge report missing targets per item, and the `AppConfigFragmentBulkWriteResult` data type carries the partial result

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

@@ -29,3 +29,23 @@ class AppConfigFragmentSearchResult:
     total_count: int
     has_next_page: bool
     has_previous_page: bool
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkItemError:
+    """One failed item of a partial bulk mutation: its batch position and a reason."""
+
+    index: int
+    message: str
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkWriteResult:
+    """Partial-success result of a bulk mutation.
+
+    ``succeeded`` are the fragments that were created/updated/purged; ``failed`` are the items
+    whose gate was rejected or whose write failed, each with its batch ``index`` and a reason.
+    """
+
+    succeeded: list[AppConfigFragmentData]
+    failed: list[AppConfigFragmentBulkItemError]

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

@@ -6,8 +6,9 @@
 from typing import Any, override
 
 from ai.backend.common.data.app_config.types import AppConfigScopeType
+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.repositories.base import CreatorSpec
+from ai.backend.manager.repositories.base import CreatorSpec, ExistsQuerier
 
 
 @dataclass
@@ -31,3 +32,15 @@ def build_row(self) -> AppConfigFragmentRow:
             scope_id=self.scope_id,
             config=self.config,
         )
+
+
+@dataclass(frozen=True)
+class GatedAppConfigFragmentCreate:
+    """One bulk-create item: a fragment spec paired with its allow-list write-gate.
+
+    Only creation is gated — update/purge of an existing fragment need no gate because
+    the FK to the allow-list guarantees the entry exists while the fragment does.
+    """
+
+    spec: AppConfigFragmentCreatorSpec
+    only_if: ExistsQuerier[AppConfigAllowListRow]

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

@@ -13,6 +13,8 @@
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
 from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkItemError,
+    AppConfigFragmentBulkWriteResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -25,10 +27,13 @@
 from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.app_config_fragment.creators import (
     AppConfigFragmentCreatorSpec,
+    GatedAppConfigFragmentCreate,
 )
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
+    BulkCreator,
     Creator,
+    CreatorSpec,
     ExistsQuerier,
     Purger,
     Querier,
@@ -58,6 +63,25 @@
 )
 
 
+def _not_found_failures(
+    pk_values: Sequence[object],
+    succeeded_ids: set[AppConfigFragmentID],
+    errored_indices: set[int],
+) -> list[AppConfigFragmentBulkItemError]:
+    """Per-item not-found errors for targets that neither succeeded nor errored.
+
+    The generic partial bulk ops silently skip a missing primary key (no row returned,
+    no error), so the missing targets are reconstructed from the input here.
+    """
+    return [
+        AppConfigFragmentBulkItemError(
+            index=index, message=f"App config fragment {pk_value} not found"
+        )
+        for index, pk_value in enumerate(pk_values)
+        if index not in errored_indices and pk_value not in succeeded_ids
+    ]
+
+
 class AppConfigFragmentDBSource:
     """Database source for app config fragment operations."""
 
@@ -112,6 +136,110 @@ async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragment
                 raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
             return result.row.to_data()
 
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_create(
+        self,
+        items: Sequence[GatedAppConfigFragmentCreate],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Create many fragments with partial success.
+
+        Every item's allow-list gate is checked in the same transaction as the writes,
+        so the gate checks and the inserts commit atomically. Rejected items are
+        reported in ``failed`` with their batch index; the allowed items are inserted
+        each in its own savepoint, so a failed insert (e.g. a duplicate natural key)
+        is also reported per item while the rest are created.
+        """
+        async with self._ops.write_ops() as w:
+            allowed: list[tuple[int, CreatorSpec[AppConfigFragmentRow]]] = []
+            failed: list[AppConfigFragmentBulkItemError] = []
+            for index, item in enumerate(items):
+                if await w.exists(item.only_if):
+                    allowed.append((index, item.spec))
+                else:
+                    failed.append(
+                        AppConfigFragmentBulkItemError(
+                            index=index,
+                            message=(
+                                f"Writing app config {item.spec.config_name!r} at scope "
+                                f"{item.spec.scope_type.value!r} is not allowed."
+                            ),
+                        )
+                    )
+            result = await w.bulk_create_partial(
+                BulkCreator(specs=[spec for _, spec in allowed])
+            )
+            failed.extend(
+                AppConfigFragmentBulkItemError(
+                    index=allowed[error.index][0], message=str(error.exception)
+                )
+                for error in result.errors
+            )
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=[row.to_data() for row in result.successes],
+                failed=sorted(failed, key=lambda e: e.index),
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_update(
+        self,
+        updaters: Sequence[Updater[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Update many fragments with partial success.
+
+        No write-gate — see ``update``. Each updater runs in its own savepoint: a
+        missing target or a failed update is reported in ``failed`` (with its batch
+        index) while the rest are updated. The batch shares one transaction, so the
+        successful updates commit together.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_update_partial(updaters)
+            succeeded = [row.to_data() for row in result.successes]
+            failed = [
+                AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                for e in result.errors
+            ]
+            failed.extend(
+                _not_found_failures(
+                    [updater.pk_value for updater in updaters],
+                    {data.id for data in succeeded},
+                    {e.index for e in result.errors},
+                )
+            )
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=succeeded,
+                failed=sorted(failed, key=lambda e: e.index),
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_purge(
+        self,
+        purgers: Sequence[Purger[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Purge many fragments with partial success.
+
+        No write-gate — see ``update``. Each purger runs in its own savepoint: a
+        missing target or a failed delete is reported in ``failed`` (with its batch
+        index) while the rest are purged.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_purge_partial(list(purgers))
+            succeeded = [row.to_data() for row in result.successes]
+            failed = [
+                AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                for e in result.errors
+            ]
+            failed.extend(
+                _not_found_failures(
+                    [purger.pk_value for purger in purgers],
+                    {data.id for data in succeeded},
+                    {e.index for e in result.errors},
+                )
+            )
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=succeeded,
+                failed=sorted(failed, key=lambda e: e.index),
+            )
+
     @app_config_fragment_db_source_resilience.apply()
     async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchResult:
         """Superadmin/internal path: query across all fragments with no scope filter."""

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

@@ -9,6 +9,7 @@
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
 from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkWriteResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -17,11 +18,17 @@
 from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.app_config_fragment.creators import (
     AppConfigFragmentCreatorSpec,
+    GatedAppConfigFragmentCreate,
 )
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
-from ai.backend.manager.repositories.base import BatchQuerier, ExistsQuerier, Purger, Updater
+from ai.backend.manager.repositories.base import (
+    BatchQuerier,
+    ExistsQuerier,
+    Purger,
+    Updater,
+)
 from ai.backend.manager.repositories.ops import DBOpsProvider
 
 __all__ = ("AppConfigFragmentRepository",)
@@ -83,3 +90,24 @@ async def scoped_search(
         self, querier: BatchQuerier, scopes: Sequence[SearchScope]
     ) -> AppConfigFragmentSearchResult:
         return await self._db_source.scoped_search(querier, scopes)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_create(
+        self,
+        items: Sequence[GatedAppConfigFragmentCreate],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_create(items)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_update(
+        self,
+        updaters: Sequence[Updater[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_update(updaters)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_purge(
+        self,
+        purgers: Sequence[Purger[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_purge(purgers)