refactor: merge admin deployment search back into the main layers

Fold DeploymentAdminRepository / DeploymentAdminService /
DeploymentAdminProcessors back into DeploymentRepository /
DeploymentService / DeploymentProcessors. admin_search_deployments now
lives alongside the scoped search methods, distinguished only by its
admin_ prefix and the absence of a scope filter — no dedicated admin
classes. Drops the now-unused DEPLOYMENT_ADMIN_REPOSITORY metric layer.

Also fix two rebase-induced CI breakages:
- main renamed execute_batch_querier's scope= kwarg to scopes=; update
  the user/project scoped search calls to scopes=[scope].
- repoint the new sokovan deploying_rolling_back handler (and its test)
  at the common DeploymentLifecycleSubStep location, since this branch
  moved the enum off manager.data.deployment.types.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: jopemachine

src/ai/backend/common/metrics/metric.py

@@ -420,7 +420,6 @@ class LayerType(enum.StrEnum):
     AUDIT_LOG_REPOSITORY = "audit_log_repository"
     CONTAINER_REGISTRY_REPOSITORY = "container_registry_repository"
     DEPLOYMENT_REPOSITORY = "deployment_repository"
-    DEPLOYMENT_ADMIN_REPOSITORY = "deployment_admin_repository"
     DOMAIN_REPOSITORY = "domain_repository"
     DOTFILE_REPOSITORY = "dotfile_repository"
     ETCD_CONFIG_REPOSITORY = "etcd_config_repository"

src/ai/backend/manager/api/adapters/deployment/adapter.py

@@ -621,7 +621,7 @@ async def admin_search(
         """Search deployments (admin, no scope)."""
         querier = self._build_deployment_querier(input)
         action_result = (
-            await self._processors.deployment_admin.admin_search_deployments.wait_for_complete(
+            await self._processors.deployment.admin_search_deployments.wait_for_complete(
                 AdminSearchDeploymentsAction(querier=querier)
             )
         )
@@ -1370,7 +1370,7 @@ async def batch_load_by_ids(
             conditions=[DeploymentConditions.by_ids(deployment_ids)],
         )
         action_result = (
-            await self._processors.deployment_admin.admin_search_deployments.wait_for_complete(
+            await self._processors.deployment.admin_search_deployments.wait_for_complete(
                 AdminSearchDeploymentsAction(querier=querier)
             )
         )

src/ai/backend/manager/repositories/deployment/__init__.py

@@ -3,11 +3,9 @@
 from ai.backend.manager.models.endpoint.conditions import DeploymentConditions
 from ai.backend.manager.models.routing.conditions import RouteConditions
 
-from .admin_repository import DeploymentAdminRepository
 from .repository import DeploymentRepository
 
 __all__ = [
-    "DeploymentAdminRepository",
     "DeploymentRepository",
     "DeploymentConditions",
     "RouteConditions",

src/ai/backend/manager/repositories/deployment/admin_repository.py

@@ -1135,7 +1135,7 @@ async def admin_search_deployments(
         """Search every endpoint without a scope filter, projecting each row
         directly to ``ModelDeploymentData``.
 
-        Backs ``DeploymentAdminRepository.admin_search_deployments`` — the
+        Backs ``DeploymentRepository.admin_search_deployments`` — the
         admin label makes the unscoped intent explicit at every layer of
         the stack (db_source → repository → service).
         """
@@ -1166,7 +1166,7 @@ async def search_user_deployments(
 
         Backs the v2 adapter's ``my_search`` path. Scope filter
         (``EndpointRow.created_user == user_id``) is injected via
-        ``execute_batch_querier``'s ``scope`` argument.
+        ``execute_batch_querier``'s ``scopes`` argument.
         """
         async with self._begin_readonly_session_read_committed() as db_sess:
             query = sa.select(EndpointRow)
@@ -1175,7 +1175,7 @@ async def search_user_deployments(
                 db_sess,
                 query,
                 querier,
-                scope=scope,
+                scopes=[scope],
             )
 
             items = [row.EndpointRow.to_model_deployment_data() for row in result.rows]
@@ -1205,7 +1205,7 @@ async def search_project_deployments(
                 db_sess,
                 query,
                 querier,
-                scope=scope,
+                scopes=[scope],
             )
 
             items = [row.EndpointRow.to_model_deployment_data() for row in result.rows]

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

@@ -5,7 +5,6 @@
 
 from ai.backend.manager.repositories.types import RepositoryArgs
 
-from .admin_repository import DeploymentAdminRepository
 from .repository import DeploymentRepository
 
 
@@ -14,7 +13,6 @@ class DeploymentRepositories:
     """Container for deployment-related repositories."""
 
     repository: DeploymentRepository
-    admin_repository: DeploymentAdminRepository
 
     @classmethod
     def create(cls, args: RepositoryArgs) -> Self:
@@ -26,5 +24,4 @@ def create(cls, args: RepositoryArgs) -> Self:
             args.valkey_live_client,
             args.valkey_schedule_client,
         )
-        admin_repository = DeploymentAdminRepository(args.db)
-        return cls(repository=repository, admin_repository=admin_repository)
+        return cls(repository=repository)

src/ai/backend/manager/repositories/deployment/repositories.py

@@ -1544,6 +1544,20 @@ async def get_route(
         """
         return await self._db_source.get_route(route_id)
 
+    @deployment_repository_resilience.apply()
+    async def admin_search_deployments(
+        self,
+        querier: BatchQuerier,
+    ) -> ModelDeploymentDataSearchResult:
+        """Search every deployment without a scope filter (admin / DataLoader).
+
+        The absence of a scope filter is what makes this admin-only; callers
+        may still narrow the result through the querier's conditions. The
+        ``admin_`` prefix is kept across db_source → repository → service so
+        the unscoped intent is obvious at every call site.
+        """
+        return await self._db_source.admin_search_deployments(querier)
+
     @deployment_repository_resilience.apply()
     async def search_user_deployments(
         self,

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

@@ -12,9 +12,10 @@
 class AdminSearchDeploymentsAction(DeploymentBaseAction):
     """Search every deployment with no scope (superadmin path).
 
-    Routed through ``DeploymentAdminProcessors`` so callers make the
-    admin intent explicit; scope-restricted variants live on the regular
-    ``DeploymentService`` and ``DeploymentRepository``.
+    The ``admin_`` prefix makes the unscoped intent explicit; scope-restricted
+    variants (``search_user_deployments`` / ``search_project_deployments``)
+    live alongside this on the same ``DeploymentService`` /
+    ``DeploymentRepository`` / ``DeploymentProcessors``.
     """
 
     querier: BatchQuerier

src/ai/backend/manager/services/deployment/actions/admin_search_deployments.py

@@ -146,7 +146,6 @@
 )
 
 if TYPE_CHECKING:
-    from ai.backend.manager.services.deployment.admin_service import DeploymentAdminService
     from ai.backend.manager.services.deployment.service import DeploymentService
 
 
@@ -167,6 +166,9 @@ class DeploymentProcessors(AbstractProcessorPackage):
     destroy_deployment: SingleEntityActionProcessor[
         DestroyDeploymentAction, DestroyDeploymentActionResult
     ]
+    admin_search_deployments: ActionProcessor[
+        AdminSearchDeploymentsAction, AdminSearchDeploymentsActionResult
+    ]
     search_user_deployments: ActionProcessor[
         SearchUserDeploymentsAction, SearchUserDeploymentsActionResult
     ]
@@ -262,6 +264,9 @@ def __init__(
         self.destroy_deployment = SingleEntityActionProcessor(
             service.destroy_deployment, action_monitors, validators=rbac_single_entity_validators
         )
+        self.admin_search_deployments = ActionProcessor(
+            service.admin_search_deployments, action_monitors
+        )
         self.search_user_deployments = ActionProcessor(
             service.search_user_deployments,
             action_monitors,
@@ -347,6 +352,7 @@ def supported_actions(self) -> list[ActionSpec]:
             UpdateDeploymentAction.spec(),
             ReplaceDeploymentOptionsAction.spec(),
             DestroyDeploymentAction.spec(),
+            AdminSearchDeploymentsAction.spec(),
             SearchUserDeploymentsAction.spec(),
             SearchProjectDeploymentsAction.spec(),
             SearchProjectDeploymentSummaryAction.spec(),
@@ -382,31 +388,3 @@ def supported_actions(self) -> list[ActionSpec]:
             BulkDeleteAccessTokensAction.spec(),
             SearchAccessTokensAction.spec(),
         ]
-
-
-class DeploymentAdminProcessors(AbstractProcessorPackage):
-    """Admin (no-scope) processors for deployments.
-
-    Counterpart of :class:`DeploymentProcessors` that owns the no-scope
-    queries — admin search and the DataLoader batch path. Scoped reads
-    stay on :class:`DeploymentProcessors`.
-    """
-
-    admin_search_deployments: ActionProcessor[
-        AdminSearchDeploymentsAction, AdminSearchDeploymentsActionResult
-    ]
-
-    def __init__(
-        self,
-        service: DeploymentAdminService,
-        action_monitors: list[ActionMonitor],
-    ) -> None:
-        self.admin_search_deployments = ActionProcessor(
-            service.admin_search_deployments, action_monitors
-        )
-
-    @override
-    def supported_actions(self) -> list[ActionSpec]:
-        return [
-            AdminSearchDeploymentsAction.spec(),
-        ]