feat(BA-5985): allow deployments without health_check via deployment_revisions.health_check_enabled flag

Persist a denormalized boolean on deployment_revisions indicating whether the
revision's model_definition declares service.health_check, computed at insert
time from ModelDefinition.is_health_check_enabled() and backfilled for
existing rows. Manager now skips HTTP probing entirely for revisions that
omit the block — routes stay NOT_CHECKED for life and are still registered
with AppProxy so traffic flows.

- Schema + ORM column + Alembic backfill (handles JSON null vs SQL NULL)
- Filter framework: RouteHealthCheckFilter dataclass; handler.health_check_filter()
  abstract classmethod; RouteTargetStatuses gains traffic_status
- Observer-dedicated repository entry: get_routes_for_health_observation()
- HealthCheckRouteHandler: skip revisions with health_check disabled
- AppProxySyncRouteHandler: include revisions with health_check disabled
  (HEALTHY OR disabled, plus traffic_status=ACTIVE) so unconfigured routes
  still register
- RouteData.session_data (replaces session_id field; session_id kept as
  property) carries SessionStatus so sync_appproxy can validate live sessions
- sync_appproxy uses input routes' replica_host/replica_port directly (same
  pattern as register_routes_now); removes the second route+kernel re-query
  via fetch_route_connection_infos and the related condition helper
- Skip routes whose session is not RUNNING/CREATING; error on missing
  session_data or replica info

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

Author: fregataa

src/ai/backend/common/config.py

@@ -489,6 +489,10 @@ 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

@@ -250,10 +250,29 @@ class DeploymentTargetStatuses:
 
 @dataclass(frozen=True)
 class RouteTargetStatuses:
-    """Target statuses for route handler filtering (lifecycle x health)."""
+    """Target statuses for route handler filtering (lifecycle x health x traffic).
+
+    ``traffic_status=None`` skips the filter; otherwise the row's
+    ``traffic_status`` must match.
+    """
 
     lifecycle: list[RouteStatus]
     health: list[RouteHealthStatus]
+    traffic_status: RouteTrafficStatus | None = None
+
+
+@dataclass(frozen=True)
+class RouteHealthCheckFilter:
+    """Revision-level ``health_check_enabled`` gating for route queries.
+
+    - ``health_check_required=None``: no filter on the flag.
+    - ``health_check_required=True/False``: AND with the flag.
+    - ``include_health_check_disabled=True``: OR-include rows where the flag
+      is ``False``, regardless of the row's ``health_status``.
+    """
+
+    health_check_required: bool | None = None
+    include_health_check_disabled: bool = False
 
 
 @dataclass(frozen=True)

src/ai/backend/manager/models/alembic/versions/a1b2c3d4e5f6_add_health_check_enabled_to_deployment_revisions.py

@@ -0,0 +1,59 @@
+"""add health_check_enabled to deployment_revisions
+
+Revision ID: a1b2c3d4e5f6
+Revises: fc249eccd0b2
+Create Date: 2026-05-09
+
+Denormalized boolean of whether the revision's ``model_definition``
+declares a ``service.health_check`` block. Backfilled by scanning
+existing JSONB rows.
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+# Part of: 26.5.0 (main)
+revision = "a1b2c3d4e5f6"
+down_revision = "fc249eccd0b2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "deployment_revisions",
+        sa.Column(
+            "health_check_enabled",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+
+    # ``->`` returns SQL NULL only when the key is absent; an explicit
+    # JSON ``null`` (which ``PydanticColumn.process_bind_param`` produces
+    # for ``health_check=None``) is a JSONB 'null' value, not SQL NULL.
+    # Filter both forms with ``jsonb_typeof(...) != 'null'``.
+    op.execute(
+        sa.text(
+            """
+            UPDATE deployment_revisions
+            SET health_check_enabled = TRUE
+            WHERE model_definition IS NOT NULL
+              AND EXISTS (
+                SELECT 1
+                FROM jsonb_array_elements(
+                    COALESCE(model_definition->'models', '[]'::jsonb)
+                ) AS m
+                WHERE m->'service'->'health_check' IS NOT NULL
+                  AND jsonb_typeof(m->'service'->'health_check') != 'null'
+              )
+            """
+        )
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("deployment_revisions", "health_check_enabled")

src/ai/backend/manager/models/deployment_revision/row.py

@@ -142,6 +142,14 @@ class DeploymentRevisionRow(Base):  # type: ignore[misc]
     model_definition: Mapped[ModelDefinition | None] = mapped_column(
         "model_definition", PydanticColumn(ModelDefinition), nullable=True
     )
+    # Set at insert time from ``model_definition.is_health_check_enabled()``.
+    health_check_enabled: Mapped[bool] = mapped_column(
+        "health_check_enabled",
+        sa.Boolean(),
+        nullable=False,
+        default=False,
+        server_default=sa.false(),
+    )
 
     # Resource configuration
     resource_group: Mapped[str] = mapped_column(

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

@@ -73,6 +73,9 @@ def build_row(self) -> DeploymentRevisionRow:
             model_mount_destination=self.model_mount_destination,
             model_definition_path=self.model_definition_path,
             model_definition=self.model_definition,
+            health_check_enabled=bool(
+                self.model_definition and self.model_definition.is_health_check_enabled()
+            ),
             resource_group=self.resource_group,
             resource_opts=self.resource_opts,
             cluster_mode=self.cluster_mode,

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

@@ -72,15 +72,16 @@
     ModelRevisionData,
     ModelRevisionSpec,
     RevisionSearchResult,
+    RouteHealthCheckFilter,
     RouteHealthStatus,
     RouteInfo,
     RouteSearchResult,
     RouteStatus,
+    RouteTargetStatuses,
     ScalingGroupCleanupConfig,
 )
 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
@@ -178,6 +179,7 @@
     ProjectDeploymentSearchScope,
     RouteData,
     RouteServiceDiscoveryInfo,
+    RouteSessionData,
 )
 from ai.backend.manager.repositories.scheduler.types.session_creation import (
     ContainerUserContext,
@@ -203,6 +205,20 @@ class EndpointWithRoutesRawData:
 log = BraceStyleAdapter(logging.getLogger(__name__))
 
 
+def _build_session_data(
+    session: uuid.UUID | None,
+    session_status: SessionStatus | None,
+) -> RouteSessionData | None:
+    """Compose ``RouteSessionData`` from a route's joined session columns.
+
+    Returns ``None`` when the route has no session yet (``session IS NULL``)
+    or when the joined session row is missing.
+    """
+    if session is None or session_status is None:
+        return None
+    return RouteSessionData(session_id=SessionId(session), status=session_status)
+
+
 def _project_preset_slots(
     preset_row: DeploymentRevisionPresetRow | None,
     slot_entries: list[tuple[str, Decimal]],
@@ -980,15 +996,17 @@ async def get_routes_by_endpoint(
     ) -> list[RouteData]:
         """Get all routes for an endpoint."""
         async with self._begin_readonly_session_read_committed() as db_sess:
-            query = sa.select(RoutingRow).where(RoutingRow.endpoint == endpoint_id)
+            query = (
+                sa.select(RoutingRow, SessionRow.status)
+                .outerjoin(SessionRow, RoutingRow.session == SessionRow.id)
+                .where(RoutingRow.endpoint == endpoint_id)
+            )
             result = await db_sess.execute(query)
-            rows = result.scalars().all()
-
             return [
                 RouteData(
                     route_id=row.id,
                     deployment_id=row.endpoint,
-                    session_id=SessionId(row.session) if row.session else None,
+                    session_data=_build_session_data(row.session, session_status),
                     status=row.status,
                     health_status=row.health_status,
                     traffic_ratio=row.traffic_ratio,
@@ -998,7 +1016,7 @@ async def get_routes_by_endpoint(
                     replica_port=row.replica_port,
                     error_data=row.error_data or {},
                 )
-                for row in rows
+                for row, session_status in result.all()
             ]
 
     async def update_route_session(
@@ -1587,24 +1605,49 @@ async def scale_routes(
 
     async def get_routes_by_statuses(
         self,
-        statuses: list[RouteStatus],
-        health_statuses: list[RouteHealthStatus],
+        target: RouteTargetStatuses,
+        health_check_filter: RouteHealthCheckFilter,
     ) -> list[RouteData]:
-        """Get routes by lifecycle and health statuses."""
+        """Routes matching ``(lifecycle, health, traffic_status)`` with
+        revision-level ``health_check_enabled`` gating per ``health_check_filter``.
+        """
         async with self._begin_readonly_session_read_committed() as db_sess:
-            query = sa.select(RoutingRow).where(
-                RoutingRow.status.in_(statuses),
-                RoutingRow.health_status.in_(health_statuses),
+            query = (
+                sa.select(
+                    RoutingRow,
+                    DeploymentRevisionRow.health_check_enabled,
+                    SessionRow.status,
+                )
+                .join(
+                    DeploymentRevisionRow,
+                    DeploymentRevisionRow.id == RoutingRow.revision,
+                )
+                .outerjoin(SessionRow, RoutingRow.session == SessionRow.id)
             )
+            query = query.where(RoutingRow.status.in_(target.lifecycle))
+            if health_check_filter.include_health_check_disabled:
+                query = query.where(
+                    sa.or_(
+                        RoutingRow.health_status.in_(target.health),
+                        DeploymentRevisionRow.health_check_enabled.is_(False),
+                    )
+                )
+            else:
+                query = query.where(RoutingRow.health_status.in_(target.health))
+            if health_check_filter.health_check_required is not None:
+                query = query.where(
+                    DeploymentRevisionRow.health_check_enabled.is_(
+                        health_check_filter.health_check_required
+                    )
+                )
+            if target.traffic_status is not None:
+                query = query.where(RoutingRow.traffic_status == target.traffic_status)
             result = await db_sess.execute(query)
-            rows: Sequence[RoutingRow] = result.scalars().all()
-
-            route_data_list: list[RouteData] = []
-            for row in rows:
-                route_data = RouteData(
+            return [
+                RouteData(
                     route_id=row.id,
                     deployment_id=row.endpoint,
-                    session_id=SessionId(row.session) if row.session else None,
+                    session_data=_build_session_data(row.session, session_status),
                     status=row.status,
                     health_status=row.health_status,
                     traffic_ratio=row.traffic_ratio,
@@ -1613,10 +1656,46 @@ async def get_routes_by_statuses(
                     replica_host=row.replica_host,
                     replica_port=row.replica_port,
                     error_data=row.error_data or {},
+                    health_check_enabled=bool(health_check_enabled),
                 )
-                route_data_list.append(route_data)
+                for row, health_check_enabled, session_status in result.all()
+            ]
 
-            return route_data_list
+    async def get_routes_for_health_observation(self) -> list[RouteData]:
+        """RUNNING routes whose revision declared ``service.health_check``."""
+        async with self._begin_readonly_session_read_committed() as db_sess:
+            query = (
+                sa.select(
+                    RoutingRow,
+                    DeploymentRevisionRow.health_check_enabled,
+                    SessionRow.status,
+                )
+                .join(
+                    DeploymentRevisionRow,
+                    DeploymentRevisionRow.id == RoutingRow.revision,
+                )
+                .outerjoin(SessionRow, RoutingRow.session == SessionRow.id)
+                .where(RoutingRow.status == RouteStatus.RUNNING)
+                .where(DeploymentRevisionRow.health_check_enabled.is_(True))
+            )
+            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_enabled=bool(health_check_enabled),
+                )
+                for row, health_check_enabled, session_status in result.all()
+            ]
 
     async def update_route_status_bulk(
         self,
@@ -2083,80 +2162,6 @@ 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``.