feat(BA-5829): convert AppConfigFragment GQL mutations to bulk-only (BEP-1052 §3)

BEP-1052 §3 specifies every write as bulk; the single-item mutations
previously exposed are replaced with the five bulk mutations below.
Internal service / repository single-row methods remain (bulk actions
loop over them with per-item transactions) so only the external GQL
surface is touched in this commit.

GQL mutations:
- `adminBulkCreateAppConfigFragments(input)` — any scope
- `adminBulkUpdateAppConfigFragments(input)` — any scope
- `adminBulkPurgeAppConfigFragments(input)` — absent keys no-oped
- `bulkCreateMyAppConfigFragments(input)` — USER + current_user, recomputed AppConfig
- `bulkUpdateMyAppConfigFragments(input)` — USER + current_user, recomputed AppConfig

Each item runs in its own transaction; failures collected per-item
in `failed: [AppConfigFragmentBulkError!]!`.

Service / Processor additions:
- `data/app_config_fragment/bulk_types.py` — per-item / error dataclasses
- `services/app_config_fragment/actions/admin_bulk_{create,update,purge}.py`
  + `bulk_{create,update}_my.py` — 5 actions + results
- `AppConfigFragmentService.admin_bulk_create` / `admin_bulk_update` /
  `admin_bulk_purge` / `bulk_create_my` / `bulk_update_my` — loop over
  admin repo; per-item try / except; my-path recomputes merged view.
- `AppConfigFragmentProcessors` wires the 5 new processors +
  `supported_actions()` extended.

DTO additions (`common/dto/manager/v2/app_config_fragment/`):
- `AdminAppConfigFragmentItemInput`, `AdminBulkCreate/Update/PurgeAppConfigFragmentsInput`
- `MyAppConfigFragmentItemInput`, `BulkCreate/UpdateMyAppConfigFragmentsInput`
- `AppConfigFragmentBulkError`, `PurgeAppConfigFragmentKey`
- `AdminBulkCreate/Update/PurgeAppConfigFragmentsPayload`

`BulkCreate/UpdateMyAppConfigFragmentsPayload` live in
`common/dto/manager/v2/app_config/response.py` next to `AppConfigNode`
to avoid a circular import (they carry recomputed `AppConfigNode`s).

Adapter additions (`api/adapters/app_config_fragment.py`):
- `admin_bulk_create` / `admin_bulk_update` / `admin_bulk_purge` /
  `my_bulk_create` / `my_bulk_update` — convert v2 DTO input →
  service bulk items → service result → v2 DTO payload.
  `my_bulk_*` resolve the current user via `current_user()`.

GQL types added under `api/gql/app_config_fragment/types/`:
- `bulk_inputs.py` — 7 input types
- `bulk_payloads.py` — 5 payload types + `AppConfigFragmentBulkErrorGQL`
  + `PurgeAppConfigFragmentKeyGQL`

Five new root mutations registered in `schema.py`; the three prior
single-item mutations (`adminCreate/Update/PurgeAppConfigFragment`)
are removed along with their resolvers.

REST bulk surface lands next in BA-5830.

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

Author: jopemachine

src/ai/backend/common/dto/manager/v2/app_config/__init__.py

@@ -11,6 +11,8 @@
 )
 from .response import (
     AppConfigNode,
+    BulkCreateMyAppConfigFragmentsPayload,
+    BulkUpdateMyAppConfigFragmentsPayload,
     GetUserAppConfigPayload,
     SearchAppConfigsPayload,
 )
@@ -26,6 +28,8 @@
     "AppConfigOrder",
     "AppConfigOrderField",
     "AppConfigScopeType",
+    "BulkCreateMyAppConfigFragmentsPayload",
+    "BulkUpdateMyAppConfigFragmentsPayload",
     "GetUserAppConfigInput",
     "GetUserAppConfigPayload",
     "OrderDirection",

src/ai/backend/common/dto/manager/v2/app_config/response.py

@@ -10,10 +10,15 @@
 from pydantic import Field
 
 from ai.backend.common.api_handlers import BaseResponseModel
-from ai.backend.common.dto.manager.v2.app_config_fragment.response import AppConfigFragmentNode
+from ai.backend.common.dto.manager.v2.app_config_fragment.response import (
+    AppConfigFragmentBulkError,
+    AppConfigFragmentNode,
+)
 
 __all__ = (
     "AppConfigNode",
+    "BulkCreateMyAppConfigFragmentsPayload",
+    "BulkUpdateMyAppConfigFragmentsPayload",
     "GetUserAppConfigPayload",
     "SearchAppConfigsPayload",
 )
@@ -55,3 +60,25 @@ class SearchAppConfigsPayload(BaseResponseModel):
     total_count: int = Field(description="Total number of AppConfigs matching the filter.")
     has_next_page: bool = Field(default=False, description="Whether there is a next page.")
     has_previous_page: bool = Field(default=False, description="Whether there is a previous page.")
+
+
+class BulkCreateMyAppConfigFragmentsPayload(BaseResponseModel):
+    """Payload for `bulkCreateMyAppConfigFragments`.
+
+    Each successfully created row produces a recomputed merged
+    `AppConfigNode`; failures are collected per-item (BEP-1052 §3).
+    """
+
+    created: list[AppConfigNode] = Field(
+        description="Recomputed merged AppConfig views for each created USER fragment.",
+    )
+    failed: list[AppConfigFragmentBulkError] = Field(description="Per-item failures.")
+
+
+class BulkUpdateMyAppConfigFragmentsPayload(BaseResponseModel):
+    """Payload for `bulkUpdateMyAppConfigFragments`."""
+
+    updated: list[AppConfigNode] = Field(
+        description="Recomputed merged AppConfig views for each updated USER fragment.",
+    )
+    failed: list[AppConfigFragmentBulkError] = Field(description="Per-item failures.")

src/ai/backend/common/dto/manager/v2/app_config_fragment/__init__.py

@@ -1,18 +1,30 @@
 from .request import (
+    AdminAppConfigFragmentItemInput,
+    AdminBulkCreateAppConfigFragmentsInput,
+    AdminBulkPurgeAppConfigFragmentsInput,
+    AdminBulkUpdateAppConfigFragmentsInput,
     AppConfigFragmentFilter,
     AppConfigFragmentKeyInput,
     AppConfigFragmentOrder,
+    BulkCreateMyAppConfigFragmentsInput,
+    BulkUpdateMyAppConfigFragmentsInput,
     CreateAppConfigFragmentInput,
+    MyAppConfigFragmentItemInput,
     PurgeAppConfigFragmentInput,
     SearchAppConfigFragmentsInput,
     UpdateAppConfigFragmentInput,
 )
 from .response import (
+    AdminBulkCreateAppConfigFragmentsPayload,
+    AdminBulkPurgeAppConfigFragmentsPayload,
+    AdminBulkUpdateAppConfigFragmentsPayload,
+    AppConfigFragmentBulkError,
     AppConfigFragmentNode,
     CreateAppConfigFragmentGQLPayload,
     CreateAppConfigFragmentPayload,
     GetAppConfigFragmentPayload,
     PurgeAppConfigFragmentGQLPayload,
+    PurgeAppConfigFragmentKey,
     PurgeAppConfigFragmentPayload,
     SearchAppConfigFragmentsPayload,
     UpdateAppConfigFragmentGQLPayload,
@@ -25,19 +37,31 @@
 )
 
 __all__ = (
+    "AdminAppConfigFragmentItemInput",
+    "AdminBulkCreateAppConfigFragmentsInput",
+    "AdminBulkCreateAppConfigFragmentsPayload",
+    "AdminBulkPurgeAppConfigFragmentsInput",
+    "AdminBulkPurgeAppConfigFragmentsPayload",
+    "AdminBulkUpdateAppConfigFragmentsInput",
+    "AdminBulkUpdateAppConfigFragmentsPayload",
+    "AppConfigFragmentBulkError",
     "AppConfigFragmentFilter",
     "AppConfigFragmentKeyInput",
     "AppConfigFragmentNode",
     "AppConfigFragmentOrder",
     "AppConfigFragmentOrderField",
     "AppConfigScopeType",
+    "BulkCreateMyAppConfigFragmentsInput",
+    "BulkUpdateMyAppConfigFragmentsInput",
     "CreateAppConfigFragmentGQLPayload",
     "CreateAppConfigFragmentInput",
     "CreateAppConfigFragmentPayload",
     "GetAppConfigFragmentPayload",
+    "MyAppConfigFragmentItemInput",
     "OrderDirection",
     "PurgeAppConfigFragmentGQLPayload",
     "PurgeAppConfigFragmentInput",
+    "PurgeAppConfigFragmentKey",
     "PurgeAppConfigFragmentPayload",
     "SearchAppConfigFragmentsInput",
     "SearchAppConfigFragmentsPayload",

src/ai/backend/common/dto/manager/v2/app_config_fragment/request.py

@@ -14,10 +14,17 @@
 from .types import AppConfigFragmentOrderField, AppConfigScopeType, OrderDirection
 
 __all__ = (
+    "AdminAppConfigFragmentItemInput",
+    "AdminBulkCreateAppConfigFragmentsInput",
+    "AdminBulkPurgeAppConfigFragmentsInput",
+    "AdminBulkUpdateAppConfigFragmentsInput",
     "AppConfigFragmentFilter",
     "AppConfigFragmentKeyInput",
     "AppConfigFragmentOrder",
+    "BulkCreateMyAppConfigFragmentsInput",
+    "BulkUpdateMyAppConfigFragmentsInput",
     "CreateAppConfigFragmentInput",
+    "MyAppConfigFragmentItemInput",
     "PurgeAppConfigFragmentInput",
     "SearchAppConfigFragmentsInput",
     "UpdateAppConfigFragmentInput",
@@ -77,6 +84,51 @@ class AppConfigFragmentOrder(BaseRequestModel):
     direction: OrderDirection = Field(default=OrderDirection.ASC, description="Order direction.")
 
 
+# ── Bulk mutation inputs (BEP-1052 §3) ──────────────────────────
+
+
+class AdminAppConfigFragmentItemInput(BaseRequestModel):
+    """Per-item input for admin bulk create / update (natural key + payload)."""
+
+    key: AppConfigFragmentKeyInput = Field(description="Natural-key identifier.")
+    extra_config: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Raw configuration payload (empty dict clears the row).",
+    )
+
+
+class AdminBulkCreateAppConfigFragmentsInput(BaseRequestModel):
+    items: list[AdminAppConfigFragmentItemInput] = Field(description="Rows to create.")
+
+
+class AdminBulkUpdateAppConfigFragmentsInput(BaseRequestModel):
+    items: list[AdminAppConfigFragmentItemInput] = Field(description="Rows to update.")
+
+
+class AdminBulkPurgeAppConfigFragmentsInput(BaseRequestModel):
+    keys: list[AppConfigFragmentKeyInput] = Field(description="Natural keys to purge.")
+
+
+class MyAppConfigFragmentItemInput(BaseRequestModel):
+    """Per-item input for self-service (`my`) bulk — `scope_type`
+    / `scope_id` are server-injected, so `name` is the only identifier.
+    """
+
+    name: str = Field(description="Policy name.")
+    extra_config: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Raw configuration payload (empty dict clears the row).",
+    )
+
+
+class BulkCreateMyAppConfigFragmentsInput(BaseRequestModel):
+    items: list[MyAppConfigFragmentItemInput] = Field(description="USER-scope rows to create.")
+
+
+class BulkUpdateMyAppConfigFragmentsInput(BaseRequestModel):
+    items: list[MyAppConfigFragmentItemInput] = Field(description="USER-scope rows to update.")
+
+
 class SearchAppConfigFragmentsInput(BaseRequestModel):
     """Input for searching fragments (raw rows) with filter / order / pagination.
 

src/ai/backend/common/dto/manager/v2/app_config_fragment/response.py

@@ -15,11 +15,16 @@
 from .types import AppConfigScopeType
 
 __all__ = (
+    "AdminBulkCreateAppConfigFragmentsPayload",
+    "AdminBulkPurgeAppConfigFragmentsPayload",
+    "AdminBulkUpdateAppConfigFragmentsPayload",
+    "AppConfigFragmentBulkError",
     "AppConfigFragmentNode",
     "CreateAppConfigFragmentGQLPayload",
     "CreateAppConfigFragmentPayload",
     "GetAppConfigFragmentPayload",
     "PurgeAppConfigFragmentGQLPayload",
+    "PurgeAppConfigFragmentKey",
     "PurgeAppConfigFragmentPayload",
     "SearchAppConfigFragmentsPayload",
     "UpdateAppConfigFragmentGQLPayload",
@@ -89,10 +94,66 @@ class PurgeAppConfigFragmentGQLPayload(BaseResponseModel):
     purged: bool = Field(description="Whether a row was actually removed.")
 
 
+# ── Bulk mutation payloads (BEP-1052 §3) ─────────────────────────
+
+
+class AppConfigFragmentBulkError(BaseResponseModel):
+    """Per-item failure information for bulk Fragment mutations."""
+
+    index: int = Field(description="Original position in the input list.")
+    scope_type: AppConfigScopeType = Field(description="Scope type of the failed row.")
+    scope_id: str = Field(description="Scope id of the failed row.")
+    name: str = Field(description="Policy name of the failed row.")
+    message: str = Field(description="Reason for the failure.")
+
+
+class PurgeAppConfigFragmentKey(BaseResponseModel):
+    """Natural-key identifier returned by bulk purge payloads."""
+
+    scope_type: AppConfigScopeType = Field(description="Scope type.")
+    scope_id: str = Field(description="Scope id.")
+    name: str = Field(description="Policy name.")
+
+
+class AdminBulkCreateAppConfigFragmentsPayload(BaseResponseModel):
+    """Payload for `adminBulkCreateAppConfigFragments`."""
+
+    created: list[AppConfigFragmentNode] = Field(description="Created fragments.")
+    failed: list[AppConfigFragmentBulkError] = Field(description="Per-item failures.")
+
+
+class AdminBulkUpdateAppConfigFragmentsPayload(BaseResponseModel):
+    """Payload for `adminBulkUpdateAppConfigFragments`."""
+
+    updated: list[AppConfigFragmentNode] = Field(description="Updated fragments.")
+    failed: list[AppConfigFragmentBulkError] = Field(description="Per-item failures.")
+
+
+class AdminBulkPurgeAppConfigFragmentsPayload(BaseResponseModel):
+    """Payload for `adminBulkPurgeAppConfigFragments`."""
+
+    purged: list[PurgeAppConfigFragmentKey] = Field(
+        description="Keys of rows actually removed (absent keys are no-oped).",
+    )
+    failed: list[AppConfigFragmentBulkError] = Field(description="Per-item failures.")
+
+
+# Bulk `my` payloads return recomputed merged views (AppConfig). To
+# avoid a circular import we declare them in `app_config/response.py`
+# and re-import them here via a forward alias below.
+
+
 class SearchAppConfigFragmentsPayload(BaseResponseModel):
     """Payload for paginated fragment search results."""
 
     items: list[AppConfigFragmentNode] = Field(description="Fragments matching the filter.")
     total_count: int = Field(description="Total number of fragments matching the filter.")
     has_next_page: bool = Field(default=False, description="Whether there is a next page.")
     has_previous_page: bool = Field(default=False, description="Whether there is a previous page.")
+
+
+# `BulkCreateMyAppConfigFragmentsPayload` / `BulkUpdateMyAppConfigFragmentsPayload`
+# return recomputed merged `AppConfig` views — they live in
+# `common/dto/manager/v2/app_config/response.py` to keep `AppConfigNode`
+# as the single source of truth and avoid a circular import from this
+# module into `app_config/response.py`.