revert: descope BA-5985 to traffic-status filter, BatchQuerier, and pagination

The original BA-5985 scope (allow deployments to omit
``service.health_check`` so the manager skips probing and routes
receive traffic immediately at RUNNING) is being moved out of this
ticket. This commit reverts the 17 commits on the branch that delivered
or scaffolded that capability — the ``deployment_revisions.health_check_enabled``
column and its GQL/DTO/adapter exposure, the ``RouteHealthCheckFilter``
variants, all hc-eligibility filtering inside route handlers, the
``RouteTargetStatuses.traffic`` field (added as opt-out scaffolding),
and the BatchQuerier-based per-handler hc-config fetch.

Net tree state equals ``origin/main``. ``changes/11534.feature.md``
is rewritten to describe the new (narrower) scope: a traffic-status
filter axis on route handlers, BatchQuerier adoption in the
coordinator's lifecycle scan, and pagination on that scan.

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

Author: fregataa

changes/11534.feature.md

@@ -1 +1 @@
-Allow model deployments to omit `service.health_check`; manager skips HTTP probing for such revisions and AppProxy receives traffic as soon as routes reach `RUNNING`.
+Route coordinator now scans lifecycle routes via `BatchQuerier` with pagination, and accepts an explicit traffic-status filter axis so handlers can target only routes whose `traffic_status` is in a given list.

src/ai/backend/common/config.py

@@ -496,10 +496,6 @@ def health_check_config(self) -> ModelHealthCheck | None:
                     return model.service.health_check
         return None
 
-    def is_health_check_enabled(self) -> bool:
-        """Whether any model in this definition declares a health_check block."""
-        return self.health_check_config() is not None
-
     def with_args_appended(self, args: list[str]) -> ModelDefinition:
         """Return a copy with ``args`` appended to each model's
         ``service.start_command`` as separate argv tokens.

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

@@ -251,15 +251,10 @@ class DeploymentTargetStatuses:
 
 @dataclass(frozen=True)
 class RouteTargetStatuses:
-    """Target statuses for route handler filtering (lifecycle x health x traffic).
-
-    ``traffic=None`` skips the filter; otherwise the row's
-    ``traffic_status`` column must be in the given list.
-    """
+    """Target statuses for route handler filtering (lifecycle x health)."""
 
     lifecycle: list[RouteStatus]
     health: list[RouteHealthStatus]
-    traffic: list[RouteTrafficStatus] | None = None
 
 
 @dataclass(frozen=True)

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

@@ -4,7 +4,7 @@
 import logging
 import uuid
 from collections import Counter, defaultdict
-from collections.abc import AsyncIterator, Collection, Mapping, Sequence
+from collections.abc import AsyncIterator, Mapping, Sequence
 from contextlib import asynccontextmanager as actxmgr
 from dataclasses import dataclass
 from datetime import UTC, datetime
@@ -79,6 +79,7 @@
 )
 from ai.backend.manager.data.deployment_revision_preset.types import ResourceSlotEntryData
 from ai.backend.manager.data.image.types import ImageIdentifier
+from ai.backend.manager.data.model_serving.types import AppProxyRouteEntry
 from ai.backend.manager.data.permission.types import RBACElementRef
 from ai.backend.manager.data.resource.types import ScalingGroupProxyTarget
 from ai.backend.manager.data.session.types import SessionStatus
@@ -980,6 +981,8 @@ async def get_routes_by_endpoint(
         async with self._begin_readonly_session_read_committed() as db_sess:
             query = sa.select(RoutingRow).where(RoutingRow.endpoint == endpoint_id)
             result = await db_sess.execute(query)
+            rows = result.scalars().all()
+
             return [
                 RouteData(
                     route_id=row.id,
@@ -994,7 +997,7 @@ async def get_routes_by_endpoint(
                     replica_port=row.replica_port,
                     error_data=row.error_data or {},
                 )
-                for row in result.scalars().all()
+                for row in rows
             ]
 
     async def update_route_session(
@@ -1581,63 +1584,38 @@ async def scale_routes(
 
     # Route operations
 
-    async def search_route_datas(
+    async def get_routes_by_statuses(
         self,
-        querier: BatchQuerier,
+        statuses: list[RouteStatus],
+        health_statuses: list[RouteHealthStatus],
     ) -> list[RouteData]:
-        """Return :class:`RouteData` rows matched by ``querier`` (no joins).
-
-        Consumers that need session status or revision-level health-check
-        config call :meth:`fetch_session_statuses_by_route_ids` or
-        :meth:`fetch_health_check_configs` separately.
-        """
+        """Get routes by lifecycle and health statuses."""
         async with self._begin_readonly_session_read_committed() as db_sess:
-            query = sa.select(RoutingRow)
-            result = await execute_batch_querier(db_sess, query, querier)
-            return [
-                RouteData(
-                    route_id=row.RoutingRow.id,
-                    deployment_id=row.RoutingRow.endpoint,
-                    session_id=SessionId(row.RoutingRow.session)
-                    if row.RoutingRow.session
-                    else None,
-                    status=row.RoutingRow.status,
-                    health_status=row.RoutingRow.health_status,
-                    traffic_ratio=row.RoutingRow.traffic_ratio,
-                    created_at=row.RoutingRow.created_at,
-                    revision_id=DeploymentRevisionID(row.RoutingRow.revision),
-                    replica_host=row.RoutingRow.replica_host,
-                    replica_port=row.RoutingRow.replica_port,
-                    error_data=row.RoutingRow.error_data or {},
-                )
-                for row in result.rows
-            ]
-
-    async def fetch_health_check_configs(
-        self,
-        revision_ids: Collection[DeploymentRevisionID],
-    ) -> Mapping[DeploymentRevisionID, ModelHealthCheck | None]:
-        """Resolve revision-level ``ModelHealthCheck`` for each revision id.
-
-        Missing revisions are omitted from the result; revisions that opted
-        out of ``service.health_check`` return ``None``.
-        """
-        if not revision_ids:
-            return {}
-        unique_ids = {uuid.UUID(str(rid)) for rid in revision_ids}
-        async with self._begin_readonly_session_read_committed() as db_sess:
-            query = sa.select(
-                DeploymentRevisionRow.id,
-                DeploymentRevisionRow.model_definition,
-            ).where(DeploymentRevisionRow.id.in_(unique_ids))
+            query = sa.select(RoutingRow).where(
+                RoutingRow.status.in_(statuses),
+                RoutingRow.health_status.in_(health_statuses),
+            )
             result = await db_sess.execute(query)
-            configs: dict[DeploymentRevisionID, ModelHealthCheck | None] = {}
-            for revision_id, model_definition in result.all():
-                config = (
-                    model_definition.health_check_config() if model_definition is not None else None
+            rows: Sequence[RoutingRow] = result.scalars().all()
+
+            route_data_list: list[RouteData] = []
+            for row in rows:
+                route_data = RouteData(
+                    route_id=row.id,
+                    deployment_id=row.endpoint,
+                    session_id=SessionId(row.session) if row.session else None,
+                    status=row.status,
+                    health_status=row.health_status,
+                    traffic_ratio=row.traffic_ratio,
+                    created_at=row.created_at,
+                    revision_id=DeploymentRevisionID(row.revision),
+                    replica_host=row.replica_host,
+                    replica_port=row.replica_port,
+                    error_data=row.error_data or {},
                 )
-                configs[DeploymentRevisionID(revision_id)] = config
-            return configs
+                route_data_list.append(route_data)
+
+            return route_data_list
 
     async def update_route_status_bulk(
         self,
@@ -1879,6 +1857,38 @@ async def update_route_replica_info(
                 )
                 await db_sess.execute(query)
 
+    async def fetch_health_check_configs_by_revision_ids(
+        self,
+        revision_ids: set[DeploymentRevisionID],
+    ) -> dict[DeploymentRevisionID, ModelHealthCheck | None]:
+        """Fetch health check configurations for revisions.
+
+        Reads only the ``model_definition`` column — variant-specific
+        health check defaults are already baked into that column at
+        revision-creation time, so no runtime dispatch by variant name
+        and no other row fields are needed. SET NULL state on
+        ``image`` / ``model`` does not affect this lookup.
+
+        Returns:
+            Mapping of revision_id to ModelHealthCheck (None if the
+            revision has no model_definition or no health_check inside).
+        """
+        if not revision_ids:
+            return {}
+
+        async with self._begin_readonly_session_read_committed() as db_sess:
+            query = sa.select(
+                DeploymentRevisionRow.id,
+                DeploymentRevisionRow.model_definition,
+            ).where(DeploymentRevisionRow.id.in_(revision_ids))
+            result = await db_sess.execute(query)
+            configs: dict[DeploymentRevisionID, ModelHealthCheck | None] = {}
+            for revision_id, model_definition in result.all():
+                configs[DeploymentRevisionID(revision_id)] = (
+                    model_definition.health_check_config() if model_definition is not None else None
+                )
+            return configs
+
     async def delete_routes_by_route_ids(
         self,
         route_ids: set[uuid.UUID],
@@ -2072,6 +2082,80 @@ async def fetch_session_statuses_by_route_ids(
 
             return status_map
 
+    async def fetch_route_connection_infos(
+        self,
+        *,
+        route_querier: BatchQuerier,
+    ) -> Mapping[uuid.UUID, list[AppProxyRouteEntry]]:
+        """Resolve routing-table entries grouped by endpoint id.
+
+        The caller composes ``route_querier`` with every filter that
+        applies (lifecycle / health / traffic_status / endpoint id set,
+        etc.) — db_source does not impose defaults and does not take a
+        separate ``endpoint_ids`` argument. The returned mapping only
+        contains endpoints that actually have at least one matching
+        route; the caller treats a missing key as "no traffic-receiving
+        routes for this endpoint" itself.
+
+        Internally fetches the filtered ``RoutingRow`` set, then bulk-
+        loads the main kernel for each running session and extracts
+        the inference port. Sessions that are not RUNNING/CREATING are
+        skipped because their kernel host:port is not stable.
+        """
+        result_map: dict[uuid.UUID, list[AppProxyRouteEntry]] = {}
+
+        async with self._begin_readonly_session_read_committed() as db_sess:
+            route_query = sa.select(RoutingRow).options(
+                selectinload(RoutingRow.session_row),
+            )
+            route_result = await execute_batch_querier(db_sess, route_query, route_querier)
+            route_rows: list[RoutingRow] = [r.RoutingRow for r in route_result.rows]
+            if not route_rows:
+                return result_map
+
+            # Only sessions whose kernel network address is stable contribute
+            # to the routing table; the rest will fall in on the next sync
+            # cycle once they reach RUNNING.
+            route_by_session: dict[uuid.UUID, RoutingRow] = {}
+            for r in route_rows:
+                if r.session is None or r.session_row is None:
+                    continue
+                if r.session_row.status not in (
+                    SessionStatus.RUNNING,
+                    SessionStatus.CREATING,
+                ):
+                    continue
+                route_by_session[r.session] = r
+
+            if not route_by_session:
+                return result_map
+
+            kernels = await KernelRow.batch_load_main_kernels_by_session_id(
+                db_sess, list(route_by_session.keys())
+            )
+
+            for kernel in kernels:
+                route = route_by_session.get(kernel.session_id)
+                if route is None or kernel.service_ports is None or not kernel.kernel_host:
+                    continue
+                # First inference port wins (legacy single-inference-port
+                # contract preserved during the row-method removal).
+                inference_port = next(
+                    (p for p in kernel.service_ports if p.get("is_inference")),
+                    None,
+                )
+                if inference_port is None or not inference_port.get("host_ports"):
+                    continue
+                entry = AppProxyRouteEntry(
+                    session_id=kernel.session_id,
+                    route_id=route.id,
+                    kernel_host=kernel.kernel_host,
+                    kernel_port=inference_port["host_ports"][0],
+                )
+                result_map.setdefault(uuid.UUID(str(route.endpoint)), []).append(entry)
+
+        return result_map
+
     async def search_deployment_ids(self, *, querier: BatchQuerier) -> list[DeploymentID]:
         """Search deployment ids using ``BatchQuerier``.
 

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

@@ -3,7 +3,7 @@
 import logging
 import uuid
 from collections import defaultdict
-from collections.abc import Collection, Mapping, Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from decimal import Decimal, DecimalException
@@ -67,12 +67,14 @@
     ModelDeploymentAutoScalingRuleData,
     ModelRevisionData,
     RevisionSearchResult,
+    RouteHealthStatus,
     RouteInfo,
     RouteSearchResult,
     RouteStatus,
     ScalingGroupCleanupConfig,
 )
 from ai.backend.manager.data.image.types import ImageIdentifier
+from ai.backend.manager.data.model_serving.types import AppProxyRouteEntry
 from ai.backend.manager.data.resource.types import ScalingGroupProxyTarget
 from ai.backend.manager.data.session.types import SessionStatus
 from ai.backend.manager.errors.service import EndpointNotFound
@@ -627,30 +629,13 @@ async def scale_routes(
     # Route operations
 
     @deployment_repository_resilience.apply()
-    async def search_route_datas(
+    async def get_routes_by_statuses(
         self,
-        querier: BatchQuerier,
+        statuses: list[RouteStatus],
+        health_statuses: list[RouteHealthStatus],
     ) -> list[RouteData]:
-        """Return :class:`RouteData` rows matched by ``querier`` (no joins).
-
-        Session status and revision-level health-check config are not
-        joined — callers fetch them via
-        :meth:`fetch_session_statuses_by_route_ids` and
-        :meth:`fetch_health_check_configs`.
-        """
-        return await self._db_source.search_route_datas(querier)
-
-    @deployment_repository_resilience.apply()
-    async def fetch_health_check_configs(
-        self,
-        revision_ids: Collection[DeploymentRevisionID],
-    ) -> Mapping[DeploymentRevisionID, ModelHealthCheck | None]:
-        """Resolve revision-level ``ModelHealthCheck`` for each id.
-
-        Revisions that opted out of ``service.health_check`` map to
-        ``None``; unknown revision ids are omitted.
-        """
-        return await self._db_source.fetch_health_check_configs(revision_ids)
+        """Get routes by lifecycle and health statuses."""
+        return await self._db_source.get_routes_by_statuses(statuses, health_statuses)
 
     @deployment_repository_resilience.apply()
     async def update_route_status_bulk(
@@ -761,6 +746,14 @@ async def update_route_replica_info(
         """Update replica_host and replica_port for routes."""
         await self._db_source.update_route_replica_info(updates)
 
+    @deployment_repository_resilience.apply()
+    async def fetch_health_check_configs_by_revision_ids(
+        self,
+        revision_ids: set[DeploymentRevisionID],
+    ) -> dict[DeploymentRevisionID, ModelHealthCheck | None]:
+        """Fetch health check configurations for revisions."""
+        return await self._db_source.fetch_health_check_configs_by_revision_ids(revision_ids)
+
     @deployment_repository_resilience.apply()
     async def delete_routes_by_route_ids(
         self,
@@ -1100,6 +1093,17 @@ async def fetch_session_statuses_by_route_ids(
         """Fetch session IDs for multiple routes."""
         return await self._db_source.fetch_session_statuses_by_route_ids(route_ids)
 
+    @deployment_repository_resilience.apply()
+    async def fetch_route_connection_infos(
+        self,
+        *,
+        route_querier: BatchQuerier,
+    ) -> Mapping[uuid.UUID, list[AppProxyRouteEntry]]:
+        """Resolve routing-table entries per endpoint via a caller-composed querier."""
+        return await self._db_source.fetch_route_connection_infos(
+            route_querier=route_querier,
+        )
+
     @deployment_repository_resilience.apply()
     async def search_deployment_ids(self, *, querier: BatchQuerier) -> list[DeploymentID]:
         """Search deployment ids using ``BatchQuerier``.