refactor: restore RouteHealthCheckFilter as in-memory post-filter on the repo

Bring back ``RouteHealthCheckFilter`` as a thin in-memory gate on the
revision's ``health_check_config`` presence, applied by
``DeploymentRepository.get_routes_by_statuses`` after the SQL query
materializes rows. The SQL surface stays a flat
``(lifecycle, health, traffic)`` selector — no widening, no JSON
predicates. Anything more specific (per-status conditional gates) lives
in the calling handler.

- ``RouteHealthCheckFilter`` exposes one knob: ``health_check_required``
  (``True``/``False``/``None``). The repo runs ``_passes_health_check``
  over each row.
- ``RouteHandler.health_check_filter()`` is abstract; every handler
  declares its filter explicitly. Lifecycle handlers return a no-op
  filter; ``HealthCheckRouteHandler`` returns ``health_check_required=True``.
- ``AppProxySyncRouteHandler`` widens ``target.health`` to
  ``[HEALTHY, NOT_CHECKED]`` so revisions that opted out of health_check
  (which stay NOT_CHECKED for life) remain candidates, and rejects the
  ``NOT_CHECKED + hc-enabled`` rows in ``execute`` — those routes are
  waiting for their first probe and must not be announced yet.
- Coordinator threads the handler's filter (and the observer's
  ``health_check_required=True``) into the repo call.
- Repo tests cover the (lifecycle x health x traffic) filter plus the
  ``health_check_required`` True/False post-filter cases. Handler tests
  assert the declared filter and (for AppProxy) the in-memory eligibility
  predicate.

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

Author: fregataa

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

@@ -261,6 +261,18 @@ class RouteTargetStatuses:
     traffic: RouteTrafficStatus | None = None
 
 
+@dataclass(frozen=True)
+class RouteHealthCheckFilter:
+    """In-memory gating on a revision's ``health_check_config``.
+
+    ``True`` keeps only routes whose revision has a config; ``False``
+    keeps only routes without; ``None`` skips the filter. Anything more
+    specific belongs in the calling handler.
+    """
+
+    health_check_required: bool | None = None
+
+
 @dataclass(frozen=True)
 class RouteTransitionTarget:
     """Target state for a route transition (lifecycle + health)."""

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

@@ -71,6 +71,7 @@
     ModelDeploymentAutoScalingRuleData,
     ModelRevisionData,
     RevisionSearchResult,
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteInfo,
     RouteSearchResult,
@@ -231,6 +232,20 @@ def _project_health_check_config(
     return model_definition.health_check_config()
 
 
+def _passes_health_check(
+    has_health_check_config: bool,
+    health_check_required: bool | None,
+) -> bool:
+    """Whether the revision's ``health_check_config`` presence matches
+    ``health_check_required`` (``None`` skips the check).
+    """
+    if health_check_required is True:
+        return has_health_check_config
+    if health_check_required is False:
+        return not has_health_check_config
+    return True
+
+
 def _project_preset_slots(
     preset_row: DeploymentRevisionPresetRow | None,
     slot_entries: list[tuple[str, Decimal]],
@@ -1618,13 +1633,15 @@ async def scale_routes(
     async def get_routes_by_statuses(
         self,
         target: RouteTargetStatuses,
+        health_check_filter: RouteHealthCheckFilter,
     ) -> list[RouteData]:
-        """Routes matching ``(lifecycle, health, traffic)``.
+        """Routes matching ``(lifecycle, health, traffic)`` with
+        revision-level ``health_check_config`` gating applied in memory.
 
         ``model_definition`` is selected so the resolved
-        ``ModelHealthCheck`` (or ``None``) can be attached to each
-        :class:`RouteData`; callers gate on ``RouteData.health_check_config``
-        in memory when revision-level health-check enablement matters.
+        ``ModelHealthCheck`` (or ``None``) is attached to each
+        :class:`RouteData`; ``health_check_filter`` then runs over those
+        materialized rows.
         """
         async with self._begin_readonly_session_read_committed() as db_sess:
             query = (
@@ -1646,23 +1663,29 @@ async def get_routes_by_statuses(
             if target.traffic is not None:
                 query = query.where(RoutingRow.traffic_status == target.traffic)
             result = await db_sess.execute(query)
-            return [
-                RouteData(
-                    route_id=row.id,
-                    deployment_id=row.endpoint,
-                    session_data=_build_session_data(row.session, session_status),
-                    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 {},
-                    health_check_config=_project_health_check_config(model_definition),
+            routes: list[RouteData] = []
+            for row, model_definition, session_status in result.all():
+                health_check_config = _project_health_check_config(model_definition)
+                has_config = health_check_config is not None
+                if not _passes_health_check(has_config, health_check_filter.health_check_required):
+                    continue
+                routes.append(
+                    RouteData(
+                        route_id=row.id,
+                        deployment_id=row.endpoint,
+                        session_data=_build_session_data(row.session, session_status),
+                        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 {},
+                        health_check_config=health_check_config,
+                    )
                 )
-                for row, model_definition, session_status in result.all()
-            ]
+            return routes
 
     async def update_route_status_bulk(
         self,

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

@@ -67,6 +67,7 @@
     ModelDeploymentAutoScalingRuleData,
     ModelRevisionData,
     RevisionSearchResult,
+    RouteHealthCheckFilter,
     RouteInfo,
     RouteSearchResult,
     RouteStatus,
@@ -631,9 +632,12 @@ async def scale_routes(
     async def get_routes_by_statuses(
         self,
         target: RouteTargetStatuses,
+        health_check_filter: RouteHealthCheckFilter,
     ) -> list[RouteData]:
-        """Routes matching ``(lifecycle, health, traffic)``."""
-        return await self._db_source.get_routes_by_statuses(target)
+        """Routes matching ``(lifecycle, health, traffic)`` with the
+        in-memory ``health_check_config`` post-filter applied.
+        """
+        return await self._db_source.get_routes_by_statuses(target, health_check_filter)
 
     @deployment_repository_resilience.apply()
     async def update_route_status_bulk(

src/ai/backend/manager/sokovan/deployment/route/coordinator.py

@@ -21,6 +21,7 @@
 from ai.backend.manager.clients.appproxy.client import AppProxyClientPool
 from ai.backend.manager.config.provider import ManagerConfigProvider
 from ai.backend.manager.data.deployment.types import (
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteStatus,
     RouteTargetStatuses,
@@ -202,7 +203,10 @@ async def process_route_lifecycle(
                 await stack.enter_async_context(self._lock_factory(handler.lock_id, lock_lifetime))
 
             target = handler.target_statuses()
-            routes = await self._deployment_repository.get_routes_by_statuses(target)
+            routes = await self._deployment_repository.get_routes_by_statuses(
+                target,
+                handler.health_check_filter(),
+            )
             if not routes:
                 log.trace("No routes to process for handler: {}", handler.name())
                 return
@@ -235,6 +239,7 @@ async def _process_observer(self, observer: RouteObserver) -> None:
                     lifecycle=[RouteStatus.RUNNING],
                     health=list(RouteHealthStatus),
                 ),
+                RouteHealthCheckFilter(health_check_required=True),
             )
             if not routes:
                 return

src/ai/backend/manager/sokovan/deployment/route/executor.py

@@ -559,9 +559,10 @@ async def sync_service_discovery(self, routes: Sequence[RouteData]) -> RouteExec
     async def sync_appproxy(self, routes: Sequence[RouteData]) -> RouteExecutionResult:
         """Push the routing tables for affected endpoints to AppProxy.
 
-        The input routes are pre-filtered by the handler (HEALTHY or
-        revision-disabled health_check), so they already represent the
-        authoritative set to register. Each route's
+        The input routes are pre-filtered by the coordinator on
+        ``(lifecycle, health, traffic)`` plus the
+        ``include_health_check_disabled`` post-filter, so they already
+        represent the authoritative set to register. Each route's
         ``replica_host``/``replica_port`` is kept up to date by
         ``check_running_routes`` and is used directly here — same pattern
         as :meth:`register_routes_now`. Routes with missing replica info

src/ai/backend/manager/sokovan/deployment/route/handlers/appproxy_sync.py

@@ -7,6 +7,7 @@
 from ai.backend.logging import BraceStyleAdapter
 from ai.backend.manager.data.deployment.types import (
     RouteHandlerCategory,
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteStatus,
     RouteStatusTransitions,
@@ -57,15 +58,18 @@ def category(cls) -> RouteHandlerCategory:
 
     @classmethod
     def target_statuses(cls) -> RouteTargetStatuses:
-        # Includes NOT_CHECKED so routes whose revision disabled health_check
-        # (they stay NOT_CHECKED forever) remain candidates; ``execute`` then
-        # narrows the in-memory set to HEALTHY or hc-disabled rows.
+        # NOT_CHECKED rows pass through for hc-disabled revisions; ``execute``
+        # rejects hc-enabled NOT_CHECKED rows whose probe is still pending.
         return RouteTargetStatuses(
             lifecycle=[RouteStatus.RUNNING],
             health=[RouteHealthStatus.HEALTHY, RouteHealthStatus.NOT_CHECKED],
             traffic=RouteTrafficStatus.ACTIVE,
         )
 
+    @classmethod
+    def health_check_filter(cls) -> RouteHealthCheckFilter:
+        return RouteHealthCheckFilter()
+
     @classmethod
     def status_transitions(cls) -> RouteStatusTransitions:
         # Pure side-effect handler; never mutates lifecycle/health columns.
@@ -76,10 +80,7 @@ def status_transitions(cls) -> RouteStatusTransitions:
         )
 
     async def execute(self, routes: Sequence[RouteData]) -> RouteExecutionResult:
-        # Register HEALTHY routes plus revisions that opted out of health_check
-        # (those carry ``health_check_config is None`` and stay NOT_CHECKED);
-        # NOT_CHECKED routes whose revision still has a probe config must wait
-        # for their first successful probe before being announced.
+        # Skip NOT_CHECKED hc-enabled rows: their first probe hasn't run yet.
         eligible = [
             r
             for r in routes

src/ai/backend/manager/sokovan/deployment/route/handlers/base.py

@@ -5,6 +5,7 @@
 
 from ai.backend.manager.data.deployment.types import (
     RouteHandlerCategory,
+    RouteHealthCheckFilter,
     RouteStatusTransitions,
     RouteTargetStatuses,
 )
@@ -59,6 +60,12 @@ def status_transitions(cls) -> RouteStatusTransitions:
         """
         raise NotImplementedError("Subclasses must implement status_transitions()")
 
+    @classmethod
+    @abstractmethod
+    def health_check_filter(cls) -> RouteHealthCheckFilter:
+        """Revision-level ``health_check_config`` gating applied by the repo."""
+        raise NotImplementedError("Subclasses must implement health_check_filter()")
+
     @abstractmethod
     async def execute(self, routes: Sequence[RouteData]) -> RouteExecutionResult:
         """Execute the route operation."""

src/ai/backend/manager/sokovan/deployment/route/handlers/health_check.py

@@ -7,6 +7,7 @@
 from ai.backend.logging import BraceStyleAdapter
 from ai.backend.manager.data.deployment.types import (
     RouteHandlerCategory,
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteStatus,
     RouteStatusTransitions,
@@ -69,19 +70,15 @@ def status_transitions(cls) -> RouteStatusTransitions:
             stale=RouteTransitionTarget(health_status=RouteHealthStatus.DEGRADED),
         )
 
-    async def execute(self, routes: Sequence[RouteData]) -> RouteExecutionResult:
-        """Execute health check for routes.
-
-        Routes whose revision opted out of ``service.health_check`` carry no
-        ``health_check_config`` and never had a ``RouteHealthRecord`` registered;
-        skipping them avoids classifying their missing record as stale/degraded.
-        """
-        eligible = [r for r in routes if r.health_check_config is not None]
-        if not eligible:
-            return RouteExecutionResult(successes=[], errors=[], stale=[])
+    @classmethod
+    def health_check_filter(cls) -> RouteHealthCheckFilter:
+        # Skip revisions with no probe — they have no RouteHealthRecord.
+        return RouteHealthCheckFilter(health_check_required=True)
 
-        log.debug("Checking health for {} routes", len(eligible))
-        return await self._route_executor.check_route_health(eligible)
+    async def execute(self, routes: Sequence[RouteData]) -> RouteExecutionResult:
+        """Execute health check for routes."""
+        log.debug("Checking health for {} routes", len(routes))
+        return await self._route_executor.check_route_health(routes)
 
     async def post_process(self, result: RouteExecutionResult) -> None:
         """Log health-check results.

src/ai/backend/manager/sokovan/deployment/route/handlers/provisioning.py

@@ -7,6 +7,7 @@
 from ai.backend.logging import BraceStyleAdapter
 from ai.backend.manager.data.deployment.types import (
     RouteHandlerCategory,
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteStatus,
     RouteStatusTransitions,
@@ -55,6 +56,10 @@ def target_statuses(cls) -> RouteTargetStatuses:
             health=list(RouteHealthStatus),
         )
 
+    @classmethod
+    def health_check_filter(cls) -> RouteHealthCheckFilter:
+        return RouteHealthCheckFilter()
+
     @classmethod
     def status_transitions(cls) -> RouteStatusTransitions:
         """Provisioning → RUNNING + NOT_CHECKED on success, FAILED_TO_START on failure."""

src/ai/backend/manager/sokovan/deployment/route/handlers/route_eviction.py

@@ -7,6 +7,7 @@
 from ai.backend.logging import BraceStyleAdapter
 from ai.backend.manager.data.deployment.types import (
     RouteHandlerCategory,
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteStatus,
     RouteStatusTransitions,
@@ -70,6 +71,10 @@ def target_statuses(cls) -> RouteTargetStatuses:
             health=list(RouteHealthStatus),
         )
 
+    @classmethod
+    def health_check_filter(cls) -> RouteHealthCheckFilter:
+        return RouteHealthCheckFilter()
+
     @classmethod
     def status_transitions(cls) -> RouteStatusTransitions:
         """Eviction: success → TERMINATING, failure → no change."""