feat(BA-6031): extend coordinator layer to support sub_status and traffic_status (#11606)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: HyeockJinKim

changes/11606.enhance.md

@@ -0,0 +1 @@
+Extend route coordinator to support sub_status and traffic_status transitions across all lifecycle handler axes.

src/ai/backend/common/dto/manager/scheduling_history/response.py

@@ -75,6 +75,8 @@ class RouteHistoryDTO(BaseResponseModel):
     phase: str
     from_status: str | None = None
     to_status: str | None = None
+    from_sub_status: str | None = None
+    to_sub_status: str | None = None
     result: str
     error_code: str | None = None
     message: str | None = None

src/ai/backend/common/dto/manager/v2/scheduling_history/response.py

@@ -75,20 +75,17 @@ class DeploymentHistoryNode(BaseResponseModel):
 
 
 class RouteHistoryNode(BaseResponseModel):
-    """Node model representing a route scheduling history record.
-
-    from_status/to_status values depend on category:
-    - category=lifecycle: lifecycle status values (provisioning, running, etc.)
-    - category=health: health status values (healthy, unhealthy, etc.)
-    """
+    """Node model representing a route scheduling history record."""
 
     id: UUID = Field(description="History record ID")
     route_id: UUID = Field(description="Route ID this history belongs to")
     deployment_id: UUID = Field(description="Deployment ID the route belongs to")
     category: str = Field(description="Handler category: 'lifecycle' or 'health'")
     phase: str = Field(description="Scheduling phase")
-    from_status: str | None = Field(default=None, description="Status before transition")
-    to_status: str | None = Field(default=None, description="Status after transition")
+    from_status: str | None = Field(default=None, description="Lifecycle status before transition")
+    to_status: str | None = Field(default=None, description="Lifecycle status after transition")
+    from_sub_status: str | None = Field(default=None, description="Sub-status before transition")
+    to_sub_status: str | None = Field(default=None, description="Sub-status after transition")
     result: str = Field(description="Result of the scheduling attempt")
     error_code: str | None = Field(default=None, description="Error code if scheduling failed")
     message: str | None = Field(default=None, description="Human-readable message or error detail")

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

@@ -760,6 +760,8 @@ def _route_data_to_dto(data: RouteHistoryData) -> RouteHistoryNode:
             phase=data.phase,
             from_status=data.from_status,
             to_status=data.to_status,
+            from_sub_status=data.from_sub_status,
+            to_sub_status=data.to_sub_status,
             result=data.result.value,
             error_code=data.error_code,
             message=data.message,

src/ai/backend/manager/api/rest/scheduling_history/adapter.py

@@ -375,6 +375,8 @@ def convert_route_history_to_dto(self, data: RouteHistoryData) -> RouteHistoryDT
             phase=data.phase,
             from_status=data.from_status,
             to_status=data.to_status,
+            from_sub_status=data.from_sub_status,
+            to_sub_status=data.to_sub_status,
             result=data.result.value,
             error_code=data.error_code,
             message=data.message,

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

@@ -265,23 +265,26 @@ class DeploymentTargetStatuses:
 
 @dataclass(frozen=True)
 class RouteTargetStatuses:
-    """Target statuses for route handler filtering (lifecycle x health x traffic).
+    """Target statuses for route handler filtering.
 
-    ``traffic=None`` skips the traffic-status predicate. Pass a non-empty list
-    to restrict to specific ``RouteTrafficStatus`` values.
+    Each axis is optional — ``None`` skips that predicate entirely.
+    Pass a non-empty list to restrict to specific values on that axis.
     """
 
-    lifecycle: list[RouteStatus]
-    health: list[RouteHealthStatus]
+    lifecycle: list[RouteStatus] | None = None
+    health: list[RouteHealthStatus] | None = None
     traffic: list[RouteTrafficStatus] | None = None
+    sub_status: list[RouteSubStatus] | None = None
 
 
 @dataclass(frozen=True)
 class RouteTransitionTarget:
-    """Target state for a route transition (lifecycle + health)."""
+    """Target state for a route transition."""
 
     status: RouteStatus | None = None
     health_status: RouteHealthStatus | None = None
+    sub_status: RouteSubStatus | None = None
+    traffic_status: RouteTrafficStatus | None = None
 
 
 @dataclass(frozen=True)
@@ -1148,21 +1151,18 @@ class DeploymentHistoryData:
 
 @dataclass
 class RouteHistoryData:
-    """Domain model for route history.
-
-    from_status/to_status contain the relevant status for the category:
-    - category=lifecycle: lifecycle status values (provisioning, running, etc.)
-    - category=health: health status values (healthy, unhealthy, etc.)
-    """
+    """Domain model for route history."""
 
     id: UUID
     route_id: UUID
     deployment_id: UUID
 
-    category: str  # RouteHandlerCategory value
+    category: RouteHandlerCategory
     phase: str  # RouteLifecycleType value
     from_status: str | None
     to_status: str | None
+    from_sub_status: str | None
+    to_sub_status: str | None
 
     result: SchedulingResult
     error_code: str | None

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

@@ -0,0 +1,37 @@
+"""Add sub_status columns to route_history and drop health_status columns.
+
+Revision ID: b2c3d4e5f6a7
+Revises: a1b2c3d4e5f7
+Create Date: 2026-05-14
+
+"""
+
+# Part of: 26.5.0
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "b2c3d4e5f6a7"
+down_revision = "d52fd86275bf"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for col in ("from_sub_status", "to_sub_status"):
+        conn.exec_driver_sql(
+            f"ALTER TABLE route_history ADD COLUMN IF NOT EXISTS {col} VARCHAR(64)"
+        )
+    for col in ("from_health_status", "to_health_status"):
+        conn.exec_driver_sql(f"ALTER TABLE route_history DROP COLUMN IF EXISTS {col}")
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for col in ("to_sub_status", "from_sub_status"):
+        op.drop_column("route_history", col)
+    for col in ("from_health_status", "to_health_status"):
+        conn.exec_driver_sql(
+            f"ALTER TABLE route_history ADD COLUMN IF NOT EXISTS {col} VARCHAR(64)"
+        )

src/ai/backend/manager/models/routing/conditions.py

@@ -10,6 +10,7 @@
 from ai.backend.manager.data.deployment.types import (
     RouteHealthStatus,
     RouteStatus,
+    RouteSubStatus,
     RouteTrafficStatus,
 )
 from ai.backend.manager.models.routing import RoutingRow
@@ -117,6 +118,13 @@ def inner() -> sa.sql.expression.ColumnElement[bool]:
 
         return inner
 
+    @staticmethod
+    def by_sub_statuses(sub_statuses: list[RouteSubStatus]) -> QueryCondition:
+        def inner() -> sa.sql.expression.ColumnElement[bool]:
+            return RoutingRow.sub_status.in_(sub_statuses)
+
+        return inner
+
     @staticmethod
     def by_revision_ids(revision_ids: Collection[uuid.UUID]) -> QueryCondition:
         def inner() -> sa.sql.expression.ColumnElement[bool]:

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

@@ -11,6 +11,7 @@
 from ai.backend.manager.data.deployment.types import (
     DeploymentHandlerCategory,
     DeploymentHistoryData,
+    RouteHandlerCategory,
     RouteHistoryData,
 )
 from ai.backend.manager.data.kernel.types import (
@@ -252,19 +253,22 @@ class RouteHistoryRow(Base):  # type: ignore[misc]
         "deployment_id", GUID, nullable=False, index=True
     )
 
-    category: Mapped[str] = mapped_column(
-        "category", sa.String(length=32), nullable=False, server_default=sa.text("'lifecycle'")
+    category: Mapped[RouteHandlerCategory] = mapped_column(
+        "category",
+        StrEnumType(RouteHandlerCategory),
+        nullable=False,
+        server_default=sa.text("'lifecycle'"),
     )
     phase: Mapped[str] = mapped_column("phase", sa.String(length=64), nullable=False)
     from_status: Mapped[str | None] = mapped_column(
         "from_status", sa.String(length=64), nullable=True
     )
     to_status: Mapped[str | None] = mapped_column("to_status", sa.String(length=64), nullable=True)
-    from_health_status: Mapped[str | None] = mapped_column(
-        "from_health_status", sa.String(length=64), nullable=True
+    from_sub_status: Mapped[str | None] = mapped_column(
+        "from_sub_status", sa.String(length=64), nullable=True
     )
-    to_health_status: Mapped[str | None] = mapped_column(
-        "to_health_status", sa.String(length=64), nullable=True
+    to_sub_status: Mapped[str | None] = mapped_column(
+        "to_sub_status", sa.String(length=64), nullable=True
     )
 
     result: Mapped[str] = mapped_column("result", sa.String(length=32), nullable=False)
@@ -296,37 +300,23 @@ class RouteHistoryRow(Base):  # type: ignore[misc]
     )
 
     def should_merge_with(self, new_row: RouteHistoryRow) -> bool:
-        """Check if a new entry should be merged with this one.
-
-        Merge conditions:
-        - Same category, phase, error_code
-        - For lifecycle category: same to_status
-        - For health category: same to_health_status
-        """
-        if self.category != new_row.category:
-            return False
-        if self.phase != new_row.phase or self.error_code != new_row.error_code:
-            return False
-        if self.category == "health":
-            return self.to_health_status == new_row.to_health_status
-        return self.to_status == new_row.to_status
+        return (
+            self.category == new_row.category
+            and self.phase == new_row.phase
+            and self.error_code == new_row.error_code
+        )
 
     def to_data(self) -> RouteHistoryData:
-        # API exposes unified from_status/to_status based on category
-        if self.category == "health":
-            from_val = self.from_health_status
-            to_val = self.to_health_status
-        else:
-            from_val = self.from_status
-            to_val = self.to_status
         return RouteHistoryData(
             id=self.id,
             route_id=self.route_id,
             deployment_id=self.deployment_id,
             category=self.category,
             phase=self.phase,
-            from_status=from_val,
-            to_status=to_val,
+            from_status=self.from_status,
+            to_status=self.to_status,
+            from_sub_status=self.from_sub_status,
+            to_sub_status=self.to_sub_status,
             result=SchedulingResult(self.result),
             error_code=self.error_code,
             message=self.message,

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

@@ -3,19 +3,21 @@
 from __future__ import annotations
 
 import uuid
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, override
 
 from ai.backend.common.identifier.deployment import DeploymentID
 from ai.backend.common.identifier.deployment_revision import DeploymentRevisionID
 from ai.backend.manager.data.deployment.types import (
     RouteHealthStatus,
     RouteStatus,
+    RouteSubStatus,
     RouteTrafficStatus,
 )
 from ai.backend.manager.models.routing import RoutingRow
 from ai.backend.manager.repositories.base import CreatorSpec
 from ai.backend.manager.repositories.base.updater import BatchUpdaterSpec
+from ai.backend.manager.types import OptionalState, TriState
 
 
 @dataclass
@@ -54,14 +56,16 @@ def build_row(self) -> RoutingRow:
 class RouteBatchUpdaterSpec(BatchUpdaterSpec[RoutingRow]):
     """BatchUpdaterSpec for batch updating routes.
 
-    Accepts optional fields and only updates fields that are specified.
-    This allows flexible partial updates for various route operations.
+    Each axis uses the appropriate optional type:
+    - :class:`OptionalState` for status fields that cannot be nullified
+    - :class:`TriState` for ``sub_status`` which must support explicit ``None``
+      (NULLIFY) when a route exits the PROVISIONING stage
     """
 
-    status: RouteStatus | None = None
-    health_status: RouteHealthStatus | None = None
-    traffic_ratio: float | None = None
-    traffic_status: RouteTrafficStatus | None = None
+    status: OptionalState[RouteStatus] = field(default_factory=OptionalState.nop)
+    health_status: OptionalState[RouteHealthStatus] = field(default_factory=OptionalState.nop)
+    traffic_status: OptionalState[RouteTrafficStatus] = field(default_factory=OptionalState.nop)
+    sub_status: TriState[RouteSubStatus] = field(default_factory=TriState.nop)
 
     @property
     @override
@@ -71,12 +75,8 @@ def row_class(self) -> type[RoutingRow]:
     @override
     def build_values(self) -> dict[str, Any]:
         values: dict[str, Any] = {}
-        if self.status is not None:
-            values["status"] = self.status
-        if self.health_status is not None:
-            values["health_status"] = self.health_status
-        if self.traffic_ratio is not None:
-            values["traffic_ratio"] = self.traffic_ratio
-        if self.traffic_status is not None:
-            values["traffic_status"] = self.traffic_status
+        self.status.update_dict(values, "status")
+        self.health_status.update_dict(values, "health_status")
+        self.traffic_status.update_dict(values, "traffic_status")
+        self.sub_status.update_dict(values, "sub_status")
         return values