refactor: model the fragment scoped search as a scope action

`ScopedSearchAppConfigFragmentAction` was a `BaseBulkAction[SearchableActionTarget]`
holding a list of domain/user targets that the repository OR-combined into one
query. A search is not a bulk mutation over entities: it acts at one RBAC scope
and returns whatever lives under it. `BaseBulkAction` exists to authorize an
operation per entity across a batch, which is why `bulk_update` / `bulk_purge`
carry the bulk RBAC validator — the scoped search carried none at all, so once an
API reaches it any authenticated caller could read any domain's or user's
fragments by naming them in the request.

Model it the way the sibling `CreateAppConfigFragmentAction` already models a
write: one `BaseScopeAction` at the single scope named by the request, deriving
its RBAC facets through `AppConfigScopeType.to_rbac_*`, wired through
`ScopeActionProcessor` with the scope RBAC validator.

The fragment row is `(scope_type, scope_id)`, so one scope expresses the query
exactly: `DomainAppConfigFragmentSearchScope` and
`UserAppConfigFragmentSearchScope` collapse into `AppConfigFragmentSearchScope`,
and `scoped_search(querier, scope)` takes a single scope.

Searching several scopes in one request is gone — it was the only thing the list
bought, and it cannot be authorized as a single scope action. Nothing consumed
the removed `queried_refs` / `element_refs()` on the result.

Tests are parametrized over the scope instead of one near-identical test per
kind; the OR-across-scopes case is dropped with the capability.

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

@@ -197,12 +197,12 @@ async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchRe
     async def scoped_search(
         self,
         querier: BatchQuerier,
-        scopes: Sequence[SearchScope],
+        scope: SearchScope,
     ) -> AppConfigFragmentSearchResult:
-        """Scoped path: query fragments restricted to ``scopes`` (combined with OR)."""
+        """Scoped path: query the fragments written at ``scope``."""
         async with self._rbac_ops_provider.read_ops() as r:
             result = await r.batch_query_with_scopes(
-                sa.select(AppConfigFragmentRow), querier, scopes
+                sa.select(AppConfigFragmentRow), querier, [scope]
             )
             return AppConfigFragmentSearchResult(
                 items=[row.AppConfigFragmentRow.to_data() for row in result.rows],

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

@@ -85,9 +85,9 @@ async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchRe
 
     @app_config_fragment_repository_resilience.apply()
     async def scoped_search(
-        self, querier: BatchQuerier, scopes: Sequence[SearchScope]
+        self, querier: BatchQuerier, scope: SearchScope
     ) -> AppConfigFragmentSearchResult:
-        return await self._db_source.scoped_search(querier, scopes)
+        return await self._db_source.scoped_search(querier, scope)
 
     @app_config_fragment_repository_resilience.apply()
     async def bulk_update(

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

@@ -9,17 +9,17 @@
 import sqlalchemy as sa
 
 from ai.backend.common.data.app_config.types import AppConfigScopeType
+from ai.backend.common.identifier.app_config import AppConfigScopeID
 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__ = (
+    "AppConfigFragmentSearchScope",
     "AppConfigScopeArguments",
     "ResolvedAppConfigScope",
-    "DomainAppConfigFragmentSearchScope",
-    "UserAppConfigFragmentSearchScope",
 )
 
 
@@ -46,48 +46,28 @@ class ResolvedAppConfigScope:
 
 
 @dataclass(frozen=True)
-class DomainAppConfigFragmentSearchScope(SearchScope):
-    """Fragments written at one domain scope (``scope_type == domain``, ``scope_id == domain_id``).
+class AppConfigFragmentSearchScope(SearchScope):
+    """The fragments written at one scope, matching the row's ``(scope_type, scope_id)``.
 
-    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.
+    ``existence_checks`` is empty: RBAC already gates whether the caller can reach the
+    scope, so a missing owner is indistinguishable from an unreachable one.
     """
 
-    domain_id: DomainID
-
-    @override
-    def to_condition(self) -> QueryCondition:
-        domain_id = self.domain_id
-
-        def inner() -> sa.sql.expression.ColumnElement[bool]:
-            return sa.and_(
-                AppConfigFragmentRow.scope_type == AppConfigScopeType.DOMAIN,
-                AppConfigFragmentRow.scope_id == domain_id,
-            )
-
-        return inner
-
-    @property
-    @override
-    def existence_checks(self) -> Sequence[ExistenceCheck[Any]]:
-        return ()
-
-
-@dataclass(frozen=True)
-class UserAppConfigFragmentSearchScope(SearchScope):
-    """Fragments written at one user scope (``scope_type == user``, ``scope_id == user_id``)."""
-
-    user_id: UserID
+    scope_type: AppConfigScopeType
+    scope_id: AppConfigScopeID | None
+    """The scope owner — ``None`` only for ``public``, which has no owner."""
 
     @override
     def to_condition(self) -> QueryCondition:
-        user_id = self.user_id
+        scope_type = self.scope_type
+        scope_id = self.scope_id
 
         def inner() -> sa.sql.expression.ColumnElement[bool]:
             return sa.and_(
-                AppConfigFragmentRow.scope_type == AppConfigScopeType.USER,
-                AppConfigFragmentRow.scope_id == user_id,
+                AppConfigFragmentRow.scope_type == scope_type,
+                AppConfigFragmentRow.scope_id.is_(None)
+                if scope_id is None
+                else AppConfigFragmentRow.scope_id == scope_id,
             )
 
         return inner

src/ai/backend/manager/services/app_config_fragment/actions/scoped_search.py

@@ -1,91 +1,68 @@
-"""Scoped app config fragment search action and its searchable targets."""
-
 from __future__ import annotations
 
-from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import override
 
-from ai.backend.common.data.permission.types import EntityType, RBACElementType
-from ai.backend.common.identifier.domain import DomainID
-from ai.backend.common.identifier.user import UserID
-from ai.backend.manager.actions.action.bulk import BaseBulkAction, BaseBulkActionResult
-from ai.backend.manager.actions.action.types import SearchableActionTarget
+from ai.backend.common.data.permission.types import RBACElementType, ScopeType
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.app_config_fragment.types import AppConfigFragmentData
 from ai.backend.manager.data.permission.types import RBACElementRef
-from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.app_config_fragment.types import (
-    DomainAppConfigFragmentSearchScope,
-    UserAppConfigFragmentSearchScope,
+    AppConfigFragmentSearchScope,
 )
 from ai.backend.manager.repositories.base import BatchQuerier
-
-
-@dataclass(frozen=True)
-class DomainAppConfigFragmentTarget(SearchableActionTarget):
-    """Scope item keyed by a domain — the fragments written at that domain scope."""
-
-    domain_id: DomainID
-
-    @override
-    def to_rbac_element_ref(self) -> RBACElementRef:
-        return RBACElementRef(element_type=RBACElementType.DOMAIN, element_id=str(self.domain_id))
-
-    @override
-    def to_search_scope(self) -> SearchScope:
-        return DomainAppConfigFragmentSearchScope(domain_id=self.domain_id)
-
-
-@dataclass(frozen=True)
-class UserAppConfigFragmentTarget(SearchableActionTarget):
-    """Scope item keyed by a user — the fragments written at that user scope."""
-
-    user_id: UserID
-
-    @override
-    def to_rbac_element_ref(self) -> RBACElementRef:
-        return RBACElementRef(element_type=RBACElementType.USER, element_id=str(self.user_id))
-
-    @override
-    def to_search_scope(self) -> SearchScope:
-        return UserAppConfigFragmentSearchScope(user_id=self.user_id)
+from ai.backend.manager.services.app_config_fragment.actions.base import (
+    AppConfigFragmentScopeAction,
+    AppConfigFragmentScopeActionResult,
+)
 
 
 @dataclass
-class ScopedSearchAppConfigFragmentAction(BaseBulkAction[SearchableActionTarget]):
-    """Scoped path: search the fragments under the given domain/user scope targets."""
+class ScopedSearchAppConfigFragmentAction(AppConfigFragmentScopeAction):
+    """Search the fragments written at one scope.
+
+    Acts at the RBAC scope named by ``scope``, so the scope RBAC validator authorizes the
+    read the same way it authorizes a write at that scope in
+    :class:`CreateAppConfigFragmentAction`.
+    """
 
-    items: list[SearchableActionTarget]
+    scope: AppConfigFragmentSearchScope
     querier: BatchQuerier
 
     @override
-    def entity_id(self) -> str | None:
-        return None
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.SEARCH
 
     @override
-    @classmethod
-    def entity_type(cls) -> EntityType:
-        return EntityType.APP_CONFIG_FRAGMENT
+    def scope_type(self) -> ScopeType:
+        return self.scope.scope_type.to_rbac_scope_type()
 
     @override
-    @classmethod
-    def operation_type(cls) -> ActionOperationType:
-        return ActionOperationType.SEARCH
+    def scope_id(self) -> str:
+        return self.scope.scope_type.to_rbac_scope_id(self.scope.scope_id)
 
     @override
-    def targets(self) -> Sequence[SearchableActionTarget]:
-        return list(self.items)
+    def target_element(self) -> RBACElementRef:
+        element = self.scope.scope_type.to_rbac_element_type()
+        if element is None:
+            return RBACElementRef(RBACElementType.APP_CONFIG_FRAGMENT, "")
+        # A non-null element type means domain or user scope, which always carries an owner.
+        return RBACElementRef(element, str(self.scope.scope_id))
 
 
 @dataclass
-class ScopedSearchAppConfigFragmentActionResult(BaseBulkActionResult):
+class ScopedSearchAppConfigFragmentActionResult(AppConfigFragmentScopeActionResult):
+    scope: AppConfigFragmentSearchScope
     data: list[AppConfigFragmentData]
     total_count: int
     has_next_page: bool
     has_previous_page: bool
-    queried_refs: list[RBACElementRef]
 
     @override
-    def element_refs(self) -> list[RBACElementRef]:
-        return list(self.queried_refs)
+    def scope_type(self) -> ScopeType:
+        return self.scope.scope_type.to_rbac_scope_type()
+
+    @override
+    def scope_id(self) -> str:
+        return self.scope.scope_type.to_rbac_scope_id(self.scope.scope_id)

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

@@ -51,7 +51,7 @@ class AppConfigFragmentProcessors(AbstractProcessorPackage):
     admin_search: ScopeActionProcessor[
         AdminSearchAppConfigFragmentAction, AdminSearchAppConfigFragmentActionResult
     ]
-    scoped_search: BulkActionProcessor[
+    scoped_search: ScopeActionProcessor[
         ScopedSearchAppConfigFragmentAction, ScopedSearchAppConfigFragmentActionResult
     ]
     update: SingleEntityActionProcessor[
@@ -84,9 +84,12 @@ def __init__(
             service.get, action_monitors, validators=[validators.rbac.single_entity]
         )
         # admin_search is system-wide (all scopes) — gated by superadmin_required at the API
-        # layer, so it carries no per-scope RBAC validator. Non-admins use scoped_search.
+        # layer, so it carries no per-scope RBAC validator. Non-admins use scoped_search,
+        # which is validated at the scope it names.
         self.admin_search = ScopeActionProcessor(service.admin_search, action_monitors)
-        self.scoped_search = BulkActionProcessor(service.scoped_search, monitors=action_monitors)
+        self.scoped_search = ScopeActionProcessor(
+            service.scoped_search, action_monitors, validators=[validators.rbac.scope]
+        )
         self.update = SingleEntityActionProcessor(
             service.update, action_monitors, validators=[validators.rbac.single_entity]
         )

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

@@ -78,15 +78,13 @@ async def admin_search(
     async def scoped_search(
         self, action: ScopedSearchAppConfigFragmentAction
     ) -> ScopedSearchAppConfigFragmentActionResult:
-        targets = list(action.targets())
-        scopes = [t.to_search_scope() for t in targets]
-        result = await self._repository.scoped_search(action.querier, scopes)
+        result = await self._repository.scoped_search(action.querier, action.scope)
         return ScopedSearchAppConfigFragmentActionResult(
+            scope=action.scope,
             data=result.items,
             total_count=result.total_count,
             has_next_page=result.has_next_page,
             has_previous_page=result.has_previous_page,
-            queried_refs=[t.to_rbac_element_ref() for t in targets],
         )
 
     async def update(