feat(BA-6555): add bulk AppConfig resolve (batch over config_names)

Add AppConfigService.resolve_bulk + ResolveBulkAppConfigAction (USER ScopeAction taking
config_names): resolve several merged AppConfigs for one (domain, user) principal in a
single query. AppConfigFragmentRepository.list_visible_fragments_bulk fetches the visible
fragments of all names at once ((config_name, rank)-ordered); the service groups by name
and deep-merges each, returning one AppConfigData per requested name in order. This is the
bulk-query path (no scoped search over merged AppConfig, per BA-6614).

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

Author: jopemachine

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

@@ -273,3 +273,33 @@ async def list_visible_fragments(
         async with self._ops.read_ops() as r:
             result = await r.batch_query_in_global(sa.select(AppConfigFragmentRow), querier)
             return [row.AppConfigFragmentRow.to_data() for row in result.rows]
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def list_visible_fragments_bulk(
+        self, config_names: list[str], scope: AppConfigResolveScope
+    ) -> list[AppConfigFragmentData]:
+        """Visible fragments for several ``config_names`` at once, in a single query.
+
+        The bulk form of :meth:`list_visible_fragments`: per name, the public OR the scope's
+        domain OR the scope's user fragment. Ordered by ``(config_name, rank)`` so the caller
+        can group by name and deep-merge each in order.
+        """
+        if not config_names:
+            return []
+        visibility = []
+        for config_name in config_names:
+            visibility.append(AppConfigFragmentConditions.by_public_visibility(config_name))
+            visibility.append(
+                AppConfigFragmentConditions.by_domain_visibility(config_name, str(scope.domain_id))
+            )
+            visibility.append(
+                AppConfigFragmentConditions.by_user_visibility(config_name, str(scope.user_id))
+            )
+        querier = BatchQuerier(
+            pagination=NoPagination(),
+            conditions=[lambda: sa.or_(*(condition() for condition in visibility))],
+            orders=[AppConfigFragmentRow.config_name.asc(), AppConfigFragmentRow.rank.asc()],
+        )
+        async with self._ops.read_ops() as r:
+            result = await r.batch_query_in_global(sa.select(AppConfigFragmentRow), querier)
+            return [row.AppConfigFragmentRow.to_data() for row in result.rows]

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

@@ -130,3 +130,9 @@ async def list_visible_fragments(
         self, config_name: str, scope: AppConfigResolveScope
     ) -> list[AppConfigFragmentData]:
         return await self._db_source.list_visible_fragments(config_name, scope)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def list_visible_fragments_bulk(
+        self, config_names: list[str], scope: AppConfigResolveScope
+    ) -> list[AppConfigFragmentData]:
+        return await self._db_source.list_visible_fragments_bulk(config_names, scope)

src/ai/backend/manager/services/app_config/actions/resolve_bulk.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import override
+
+from ai.backend.common.data.permission.types import RBACElementType, ScopeType
+from ai.backend.common.identifier.domain import DomainID
+from ai.backend.common.identifier.user import UserID
+from ai.backend.manager.actions.types import ActionOperationType
+from ai.backend.manager.data.app_config.types import AppConfigData
+from ai.backend.manager.data.permission.types import RBACElementRef
+from ai.backend.manager.services.app_config.actions.base import (
+    AppConfigScopeAction,
+    AppConfigScopeActionResult,
+)
+
+
+@dataclass
+class ResolveBulkAppConfigAction(AppConfigScopeAction):
+    """Resolve several merged ``AppConfig``s for one ``(user, domain)`` principal at once.
+
+    A batch over ``config_names`` for a single principal (one ``USER`` scope) — not a
+    scoped search. Each name is merged independently and returned in request order.
+    """
+
+    config_names: list[str]
+    domain_id: DomainID
+    user_id: UserID
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.GET
+
+    @override
+    def scope_type(self) -> ScopeType:
+        return ScopeType.USER
+
+    @override
+    def scope_id(self) -> str:
+        return str(self.user_id)
+
+    @override
+    def target_element(self) -> RBACElementRef:
+        return RBACElementRef(RBACElementType.APP_CONFIG, "")
+
+
+@dataclass
+class ResolveBulkAppConfigActionResult(AppConfigScopeActionResult):
+    app_configs: list[AppConfigData]
+    user_id: UserID
+
+    @override
+    def scope_type(self) -> ScopeType:
+        return ScopeType.USER
+
+    @override
+    def scope_id(self) -> str:
+        return str(self.user_id)

src/ai/backend/manager/services/app_config/processors.py

@@ -9,21 +9,28 @@
     ResolveAppConfigAction,
     ResolveAppConfigActionResult,
 )
+from ai.backend.manager.services.app_config.actions.resolve_bulk import (
+    ResolveBulkAppConfigAction,
+    ResolveBulkAppConfigActionResult,
+)
 from ai.backend.manager.services.app_config.service import AppConfigService
 
 
 class AppConfigProcessors(AbstractProcessorPackage):
     resolve: ScopeActionProcessor[ResolveAppConfigAction, ResolveAppConfigActionResult]
+    resolve_bulk: ScopeActionProcessor[ResolveBulkAppConfigAction, ResolveBulkAppConfigActionResult]
 
     def __init__(
         self,
         service: AppConfigService,
         action_monitors: list[ActionMonitor],
     ) -> None:
         self.resolve = ScopeActionProcessor(service.resolve, action_monitors)
+        self.resolve_bulk = ScopeActionProcessor(service.resolve_bulk, action_monitors)
 
     @override
     def supported_actions(self) -> list[ActionSpec]:
         return [
             ResolveAppConfigAction.spec(),
+            ResolveBulkAppConfigAction.spec(),
         ]

src/ai/backend/manager/services/app_config/service.py

@@ -14,6 +14,10 @@
     ResolveAppConfigAction,
     ResolveAppConfigActionResult,
 )
+from ai.backend.manager.services.app_config.actions.resolve_bulk import (
+    ResolveBulkAppConfigAction,
+    ResolveBulkAppConfigActionResult,
+)
 
 __all__ = ("AppConfigService",)
 
@@ -59,3 +63,29 @@ async def resolve(self, action: ResolveAppConfigAction) -> ResolveAppConfigActio
             config=self._merge_configs(fragments),
         )
         return ResolveAppConfigActionResult(app_config=app_config, user_id=action.user_id)
+
+    async def resolve_bulk(
+        self, action: ResolveBulkAppConfigAction
+    ) -> ResolveBulkAppConfigActionResult:
+        """Resolve several merged ``AppConfig``s for one principal in a single query.
+
+        Fetches the visible fragments of all ``config_names`` at once, groups them by name,
+        and deep-merges each in ``rank`` order. Returns one ``AppConfigData`` per requested
+        name, in request order (an unregistered name yields ``config = None``).
+        """
+        scope = AppConfigResolveScope(domain_id=action.domain_id, user_id=action.user_id)
+        fragments = await self._fragment_repository.list_visible_fragments_bulk(
+            action.config_names, scope
+        )
+        grouped: dict[str, list[AppConfigFragmentData]] = {}
+        for fragment in fragments:
+            grouped.setdefault(fragment.config_name, []).append(fragment)
+        app_configs = [
+            AppConfigData(
+                config_name=config_name,
+                fragments=grouped.get(config_name, []),
+                config=self._merge_configs(grouped.get(config_name, [])),
+            )
+            for config_name in action.config_names
+        ]
+        return ResolveBulkAppConfigActionResult(app_configs=app_configs, user_id=action.user_id)

tests/unit/manager/repositories/app_config_fragment/test_repository.py

@@ -855,3 +855,38 @@ async def test_unknown_config_name_returns_empty(
             AppConfigResolveScope(domain_id=DomainID(_DOMAIN_UUID), user_id=UserID(_USER_UUID)),
         )
         assert applicable == []
+
+    async def test_bulk_returns_visible_fragments_for_all_names_ordered(
+        self,
+        repository: AppConfigFragmentRepository,
+        fragments_across_scopes: list[AppConfigFragmentData],
+    ) -> None:
+        applicable = await repository.list_visible_fragments_bulk(
+            ["theme", "menu"],
+            AppConfigResolveScope(domain_id=DomainID(_DOMAIN_UUID), user_id=UserID(_USER_UUID)),
+        )
+        # public + the caller's domain + the caller's own user fragment, for both names.
+        expected = {
+            f.id
+            for f in fragments_across_scopes
+            if f.config_name in ("theme", "menu")
+            and (
+                f.scope_type is AppConfigScopeType.PUBLIC
+                or (f.scope_type is AppConfigScopeType.DOMAIN and f.scope_id == _DOMAIN_ID)
+                or (f.scope_type is AppConfigScopeType.USER and f.scope_id == _USER_ID)
+            )
+        }
+        assert {f.id for f in applicable} == expected
+        # Grouped by config_name, rank-ordered within each.
+        assert [f.config_name for f in applicable] == sorted(f.config_name for f in applicable)
+
+    async def test_bulk_empty_names_returns_empty(
+        self,
+        repository: AppConfigFragmentRepository,
+        fragments_across_scopes: list[AppConfigFragmentData],
+    ) -> None:
+        applicable = await repository.list_visible_fragments_bulk(
+            [],
+            AppConfigResolveScope(domain_id=DomainID(_DOMAIN_UUID), user_id=UserID(_USER_UUID)),
+        )
+        assert applicable == []

tests/unit/manager/services/app_config/test_service.py

@@ -8,18 +8,21 @@
 
 import pytest
 
+from ai.backend.common.data.app_config.types import AppConfigScopeType
 from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
 from ai.backend.common.identifier.domain import DomainID
 from ai.backend.common.identifier.user import UserID
 from ai.backend.manager.data.app_config_fragment.types import (
     AppConfigFragmentData,
-    AppConfigScopeType,
 )
 from ai.backend.manager.repositories.app_config_fragment.repository import (
     AppConfigFragmentRepository,
 )
 from ai.backend.manager.repositories.app_config_fragment.types import AppConfigResolveScope
 from ai.backend.manager.services.app_config.actions.resolve import ResolveAppConfigAction
+from ai.backend.manager.services.app_config.actions.resolve_bulk import (
+    ResolveBulkAppConfigAction,
+)
 from ai.backend.manager.services.app_config.service import AppConfigService
 
 _USER_UUID = uuid.uuid4()
@@ -28,12 +31,17 @@
 
 
 def _fragment(
-    *, scope_type: AppConfigScopeType, scope_id: str, rank: int, config: dict[str, object]
+    *,
+    scope_type: AppConfigScopeType,
+    scope_id: str,
+    rank: int,
+    config: dict[str, object],
+    config_name: str = "theme",
 ) -> AppConfigFragmentData:
     now = datetime.now(UTC)
     return AppConfigFragmentData(
         id=AppConfigFragmentID(uuid.uuid4()),
-        config_name="theme",
+        config_name=config_name,
         scope_type=scope_type,
         scope_id=scope_id,
         rank=rank,
@@ -93,3 +101,51 @@ async def test_resolve_empty_projects_config_to_none(
 
         assert result.app_config.fragments == []
         assert result.app_config.config is None
+
+    async def test_resolve_bulk_groups_by_name_and_merges_each(
+        self, service: AppConfigService, mock_fragment_repository: MagicMock
+    ) -> None:
+        # Repo returns visible fragments for both names, (config_name, rank)-ordered.
+        theme_public = _fragment(
+            config_name="theme",
+            scope_type=AppConfigScopeType.PUBLIC,
+            scope_id="public",
+            rank=100,
+            config={"theme": "light", "lang": "en"},
+        )
+        theme_user = _fragment(
+            config_name="theme",
+            scope_type=AppConfigScopeType.USER,
+            scope_id=str(_USER_ID),
+            rank=300,
+            config={"theme": "dark"},
+        )
+        menu_public = _fragment(
+            config_name="menu",
+            scope_type=AppConfigScopeType.PUBLIC,
+            scope_id="public",
+            rank=100,
+            config={"items": ["a"]},
+        )
+        mock_fragment_repository.list_visible_fragments_bulk = AsyncMock(
+            return_value=[theme_public, theme_user, menu_public]
+        )
+
+        result = await service.resolve_bulk(
+            ResolveBulkAppConfigAction(
+                config_names=["theme", "menu", "unknown"],
+                domain_id=_DOMAIN_ID,
+                user_id=_USER_ID,
+            )
+        )
+
+        # One AppConfigData per requested name, in request order.
+        assert [c.config_name for c in result.app_configs] == ["theme", "menu", "unknown"]
+        assert result.app_configs[0].config == {"theme": "dark", "lang": "en"}
+        assert result.app_configs[1].config == {"items": ["a"]}
+        assert result.app_configs[2].config is None  # unregistered -> empty merge
+        assert result.user_id == _USER_ID
+        mock_fragment_repository.list_visible_fragments_bulk.assert_called_once_with(
+            ["theme", "menu", "unknown"],
+            AppConfigResolveScope(domain_id=_DOMAIN_ID, user_id=_USER_ID),
+        )