feat(BA-6555): add app_config merge engine (deep-merge across visible fragments)

Adds the read-side AppConfigService.resolve: fetches the fragments visible to a
(domain, user) principal for one config_name (list_visible_fragments, via the
by_*_visibility conditions) and deep-merges them in rank order; AppConfigResolveScope
bundles the principal. Rebased onto the domain/user-keyed scoped_search (BA-6554):
ConfigNameSearchScope is gone, so the merged read relies on the visibility conditions.

Squash of the BA-6555 development history, re-based onto the updated BA-6554.

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

Author: jopemachine

changes/12359.feature.md

@@ -0,0 +1 @@
+Add the app_config service layer: resolve a user's merged AppConfig for a config name by rank-ordering the applicable public / domain / user fragments and deep-merging them.

src/ai/backend/manager/data/app_config/__init__.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from ai.backend.manager.data.app_config_fragment.types import AppConfigFragmentData
+
+
+@dataclass(frozen=True)
+class AppConfigData:
+    """Merged per-user view of one ``config_name`` (BEP-1052).
+
+    The ordered contributing ``fragments`` (rank low -> high) plus their deep-merged
+    ``config``. ``config`` is ``None`` when the merge of every fragment is empty.
+    """
+
+    config_name: str
+    fragments: list[AppConfigFragmentData]
+    config: dict[str, Any] | None

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

@@ -111,6 +111,46 @@ def inner() -> sa.sql.expression.ColumnElement[bool]:
 
         return inner
 
+    # --- per-scope visibility filters (one scope_type each) for a config_name ---
+
+    @staticmethod
+    def by_public_visibility(config_name: str) -> QueryCondition:
+        """The ``public`` fragment of ``config_name`` (public has no per-entity scope_id)."""
+
+        def inner() -> sa.sql.expression.ColumnElement[bool]:
+            return sa.and_(
+                AppConfigFragmentRow.config_name == config_name,
+                AppConfigFragmentRow.scope_type == AppConfigScopeType.PUBLIC,
+            )
+
+        return inner
+
+    @staticmethod
+    def by_domain_visibility(config_name: str, domain: str) -> QueryCondition:
+        """The ``domain`` fragment of ``config_name`` for ``domain``."""
+
+        def inner() -> sa.sql.expression.ColumnElement[bool]:
+            return sa.and_(
+                AppConfigFragmentRow.config_name == config_name,
+                AppConfigFragmentRow.scope_type == AppConfigScopeType.DOMAIN,
+                AppConfigFragmentRow.scope_id == domain,
+            )
+
+        return inner
+
+    @staticmethod
+    def by_user_visibility(config_name: str, user_id: str) -> QueryCondition:
+        """The ``user`` fragment of ``config_name`` for ``user_id``."""
+
+        def inner() -> sa.sql.expression.ColumnElement[bool]:
+            return sa.and_(
+                AppConfigFragmentRow.config_name == config_name,
+                AppConfigFragmentRow.scope_type == AppConfigScopeType.USER,
+                AppConfigFragmentRow.scope_id == user_id,
+            )
+
+        return inner
+
     # --- created_at / updated_at datetime filters ---
 
     @staticmethod

src/ai/backend/manager/models/app_config_fragment/conditions.py

@@ -23,18 +23,23 @@
     AppConfigFragmentWriteNotAllowed,
 )
 from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
+from ai.backend.manager.models.app_config_fragment.conditions import AppConfigFragmentConditions
 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.app_config_fragment.types import (
+    AppConfigResolveScope,
+)
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
     BulkConditionalCreator,
     BulkConditionalPurger,
     BulkConditionalUpdater,
     Creator,
     ExistsQuerier,
+    NoPagination,
     Purger,
     Querier,
     Updater,
@@ -245,3 +250,26 @@ async def scoped_search(
                 has_next_page=result.has_next_page,
                 has_previous_page=result.has_previous_page,
             )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def list_visible_fragments(
+        self, config_name: str, scope: AppConfigResolveScope
+    ) -> list[AppConfigFragmentData]:
+        """Fragments visible to ``scope`` for ``config_name``, ``rank``-ordered, in one query.
+
+        Filters on the fragment's own ``scope_type``/``scope_id`` columns: the public OR the
+        scope's domain OR the scope's user fragment. Each ``(config_name, scope_type,
+        scope_id)`` is unique, so the result is bounded (one per scope_type), ready to
+        deep-merge in order.
+        """
+        public = AppConfigFragmentConditions.by_public_visibility(config_name)
+        domain = AppConfigFragmentConditions.by_domain_visibility(config_name, str(scope.domain_id))
+        user = AppConfigFragmentConditions.by_user_visibility(config_name, str(scope.user_id))
+        querier = BatchQuerier(
+            pagination=NoPagination(),
+            conditions=[lambda: sa.or_(public(), domain(), user())],
+            orders=[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/db_source/db_source.py

@@ -22,6 +22,9 @@
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
+from ai.backend.manager.repositories.app_config_fragment.types import (
+    AppConfigResolveScope,
+)
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
     BulkConditionalCreator,
@@ -121,3 +124,9 @@ async def bulk_purge(
         bulk: BulkConditionalPurger[AppConfigFragmentRow, AppConfigAllowListRow],
     ) -> AppConfigFragmentBulkWriteResult:
         return await self._db_source.bulk_purge(bulk)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def list_visible_fragments(
+        self, config_name: str, scope: AppConfigResolveScope
+    ) -> list[AppConfigFragmentData]:
+        return await self._db_source.list_visible_fragments(config_name, scope)

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

@@ -1,4 +1,4 @@
-"""Types for app config fragment repository operations (search scopes, gated writes)."""
+"""Types for app config fragment repository operations (search scopes, resolve scope)."""
 
 from __future__ import annotations
 
@@ -16,11 +16,25 @@
 from ai.backend.manager.models.scopes import ExistenceCheck, SearchScope
 
 __all__ = (
+    "AppConfigResolveScope",
     "DomainAppConfigFragmentSearchScope",
     "UserAppConfigFragmentSearchScope",
 )
 
 
+@dataclass(frozen=True)
+class AppConfigResolveScope:
+    """The principal an ``AppConfig`` is resolved for: the resolving user and its domain.
+
+    Bundles the scope-identifying arguments so they travel together (add new principal
+    dimensions here rather than growing method signatures). Plain value object — not a
+    :class:`SearchScope`.
+    """
+
+    domain_id: DomainID
+    user_id: UserID
+
+
 @dataclass(frozen=True)
 class DomainAppConfigFragmentSearchScope(SearchScope):
     """Fragments written at one domain scope (``scope_type == domain``, ``scope_id == domain_id``).

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

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from typing import override
+
+from ai.backend.common.data.permission.types import EntityType
+from ai.backend.manager.actions.action.scope import BaseScopeAction, BaseScopeActionResult
+
+
+class AppConfigScopeAction(BaseScopeAction):
+    """Base for scope-level merged app config actions (resolve)."""
+
+    @override
+    @classmethod
+    def entity_type(cls) -> EntityType:
+        return EntityType.APP_CONFIG
+
+
+class AppConfigScopeActionResult(BaseScopeActionResult):
+    pass