refactor(BA-5650-I): test and remaining ORM updates

Final slice of the BA-5650 stack split. Contains:

- Remaining ORM touch-ups: models/endpoint/row.py,
  models/keypair/row.py, repositories/scheduler/{repository,db_source}.py,
  api/adapters/vfolder.py, api/gql_legacy/endpoint.py,
  api/gql_legacy/routing.py.
- Test updates that depend on the action/service/DTO renames already
  landed in earlier slices: adapter/session, scheduler repositories,
  session lifecycle/service, sokovan scheduler suite,
  compute_sessions handler, dependency injection tests.
- Autouse ``_user_context`` fixture under
  tests/unit/manager/services/session/conftest.py so service tests work
  without the auth middleware.

Author: jopemachine

changes/10916.breaking.md

@@ -0,0 +1 @@
+Change the REST v1 session API's delegation mechanism from the `owner_access_key` query parameter to an `owner_id` (user UUID) field. The `owner_access_key` parameter is removed from all session endpoints (`GET /session/{name}`, `DELETE /session/{name}`, `POST /session/_/create-from-template`, `POST /session/_/create`, `POST /session/_/create-cluster`, `GET /session/{name}/logs`, `GET /session/{name}/status-history`, etc.). Clients that previously passed `owner_access_key=<keypair>` to act on behalf of another user must now pass `owner_id=<user uuid>` (and only on the session-creation endpoints; for read/control endpoints the caller always acts as themselves). Session and kernel `access_key` semantics also change: the value is no longer tied to the keypair used to create the session but resolved at read time from the owner's `main_access_key`.

changes/BA-5650-I.misc.md

@@ -0,0 +1 @@
+Test updates and remaining ORM/repository touch-ups for the BA-5650 stack: scheduler db_source, keypair and endpoint row cleanup, gql_legacy endpoint/routing, session lifecycle/service tests, sokovan scheduler tests, and dependency-injection tests. No external behavior change beyond what earlier slices documented.

dev

@@ -149,32 +149,14 @@ cmd_restart() {
 
 cmd_log() {
   local svc=$1
-  local follow=${2:-}
   local winname
   winname=$(_tmux_window_name "$svc")
   local win
   win=$(tmux list-windows -t "$TMUX_SESSION" -F "#{window_name}" 2>/dev/null | grep "^${winname}$" | head -1) || true
-  if [ -z "$win" ]; then
-    echo "$(_color red "No tmux window found for $svc")"
-    return 1
-  fi
-  if [ "$follow" = "-f" ]; then
-    local last_hash=""
-    trap 'exit 0' INT
-    while true; do
-      local output
-      output=$(tmux capture-pane -t "$TMUX_SESSION:$win" -p -S -50 2>/dev/null)
-      local cur_hash
-      cur_hash=$(echo "$output" | md5sum | cut -d' ' -f1)
-      if [ "$cur_hash" != "$last_hash" ]; then
-        clear
-        echo "$output"
-        last_hash=$cur_hash
-      fi
-      sleep 1
-    done
-  else
+  if [ -n "$win" ]; then
     tmux capture-pane -t "$TMUX_SESSION:$win" -p -S -50
+  else
+    echo "$(_color red "No tmux window found for $svc")"
   fi
 }
 
@@ -188,7 +170,7 @@ Commands:
   start   <service|all>     Start a service
   stop    <service|all>     Stop a service
   restart <service|all>     Restart a service
-  log     <service> [-f]    Show recent log output (-f to follow)
+  log     <service>         Show recent log output
 
 Services:
   mgr, agent, storage, web, proxy-coordinator, proxy-worker, all
@@ -225,9 +207,9 @@ case "$1" in
     "cmd_$1" "$2"
     ;;
   log)
-    [ $# -lt 2 ] && { echo "$(_color red "Usage: ./dev log <service> [-f]")"; exit 1; }
+    [ $# -lt 2 ] && { echo "$(_color red "Usage: ./dev log <service>")"; exit 1; }
     _validate_service "$2"
-    cmd_log "$2" "${3:-}"
+    cmd_log "$2"
     ;;
   *) echo "$(_color red "Unknown command: $1")"; usage; exit 1 ;;
 esac

docs/manager/graphql-reference/schema.graphql

@@ -1836,9 +1836,6 @@ type Routing implements Item {
   endpoint: String
   session: UUID
   status: String
-
-  """Added in 26.4.1."""
-  health_status: String
   traffic_ratio: Float
   created_at: DateTime
   error_data: JSONString

docs/manager/graphql-reference/supergraph.graphql

@@ -166,16 +166,6 @@ input AddRevisionInput
   extraMounts: [ExtraVFolderMountInput!] = null
 }
 
-"""Added in 26.4.1. Options for the add_model_revision mutation."""
-input AddRevisionOptions
-  @join__type(graph: STRAWBERRY)
-{
-  """
-  When true, automatically activate the newly added revision immediately after creation.
-  """
-  autoActivate: Boolean! = false
-}
-
 """Added in 25.19.0. Payload for adding a revision."""
 type AddRevisionPayload
   @join__type(graph: STRAWBERRY)
@@ -10019,7 +10009,7 @@ type Mutation
   syncReplicas(input: SyncReplicaInput!): SyncReplicaPayload! @join__field(graph: STRAWBERRY)
 
   """Added in 25.16.0. Add model revision."""
-  addModelRevision(input: AddRevisionInput!, options: AddRevisionOptions = null): AddRevisionPayload! @join__field(graph: STRAWBERRY)
+  addModelRevision(input: AddRevisionInput!): AddRevisionPayload! @join__field(graph: STRAWBERRY)
 
   """
   Added in 26.4.1. Create or update the deployment policy for a given deployment (upsert semantics). If the deployment already has a policy, it is replaced entirely with the new configuration
@@ -14933,9 +14923,6 @@ type Routing implements Item
   endpoint: String
   session: UUID
   status: String
-
-  """Added in 26.4.1."""
-  health_status: String
   traffic_ratio: Float
   created_at: DateTime
   error_data: JSONString

docs/manager/graphql-reference/v2-schema.graphql

@@ -126,14 +126,6 @@ input AddRevisionInput {
   extraMounts: [ExtraVFolderMountInput!] = null
 }
 
-"""Added in 26.4.1. Options for the add_model_revision mutation."""
-input AddRevisionOptions {
-  """
-  When true, automatically activate the newly added revision immediately after creation.
-  """
-  autoActivate: Boolean! = false
-}
-
 """Added in 25.19.0. Payload for adding a revision."""
 type AddRevisionPayload {
   """Added revision"""
@@ -6021,7 +6013,7 @@ type Mutation {
   syncReplicas(input: SyncReplicaInput!): SyncReplicaPayload!
 
   """Added in 25.16.0. Add model revision."""
-  addModelRevision(input: AddRevisionInput!, options: AddRevisionOptions = null): AddRevisionPayload!
+  addModelRevision(input: AddRevisionInput!): AddRevisionPayload!
 
   """
   Added in 26.4.1. Create or update the deployment policy for a given deployment (upsert semantics). If the deployment already has a policy, it is replaced entirely with the new configuration

src/ai/backend/common/clients/prometheus/__init__.py

@@ -1,11 +1,9 @@
 from .client import PrometheusClient
-from .preset import LabelMatcher, LabelOperator, MetricPreset
+from .preset import MetricPreset
 from .querier import ContainerMetricQuerier, MetricQuerier
 from .types import ValueType
 
 __all__ = [
-    "LabelMatcher",
-    "LabelOperator",
     "PrometheusClient",
     "MetricPreset",
     "MetricQuerier",

src/ai/backend/common/clients/prometheus/preset.py

@@ -1,30 +1,5 @@
 from collections.abc import Mapping, Set
 from dataclasses import dataclass, field
-from enum import StrEnum
-from typing import Self
-
-
-class LabelOperator(StrEnum):
-    EQUAL = "="
-    NOT_EQUAL = "!="
-    REGEX = "=~"
-    NOT_REGEX = "!~"
-
-
-@dataclass(frozen=True)
-class LabelMatcher:
-    """PromQL label matcher with an explicit operator."""
-
-    value: str
-    operator: LabelOperator = LabelOperator.EQUAL
-
-    @classmethod
-    def exact(cls, value: str) -> Self:
-        return cls(value=value, operator=LabelOperator.EQUAL)
-
-    @classmethod
-    def regex(cls, value: str) -> Self:
-        return cls(value=value, operator=LabelOperator.REGEX)
 
 
 def _escape_label_value(value: str) -> str:
@@ -40,7 +15,7 @@ class MetricPreset:
     template: str
 
     # Query labels (injected into {labels} placeholder)
-    labels: Mapping[str, LabelMatcher] = field(default_factory=dict)
+    labels: Mapping[str, str] = field(default_factory=dict)
 
     # Group by labels (injected into {group_by} placeholder)
     group_by: Set[str] = field(default_factory=frozenset)
@@ -50,10 +25,7 @@ class MetricPreset:
 
     def render(self) -> str:
         """Render the PromQL query with all values injected."""
-        label_str = ",".join(
-            f'{key}{value.operator}"{_escape_label_value(value.value)}"'
-            for key, value in self.labels.items()
-        )
+        label_str = ",".join(f'{k}="{_escape_label_value(v)}"' for k, v in self.labels.items())
         return self.template.format(
             labels=label_str,
             window=self.window,

src/ai/backend/common/clients/prometheus/querier.py

@@ -3,7 +3,6 @@
 from dataclasses import dataclass
 from uuid import UUID
 
-from ai.backend.common.clients.prometheus.preset import LabelMatcher
 from ai.backend.common.clients.prometheus.types import ValueType
 
 
@@ -15,7 +14,7 @@ class MetricQuerier(ABC):
     """
 
     @abstractmethod
-    def labels(self) -> Mapping[str, LabelMatcher]:
+    def labels(self) -> Mapping[str, str]:
         """Return the labels to be used in the Prometheus query."""
         ...
 
@@ -35,22 +34,22 @@ class ContainerMetricQuerier(MetricQuerier):
     user_id: UUID | None = None
     project_id: UUID | None = None
 
-    def labels(self) -> Mapping[str, LabelMatcher]:
+    def labels(self) -> Mapping[str, str]:
         """Return the labels for the container metric query."""
-        result: dict[str, LabelMatcher] = {
-            "container_metric_name": LabelMatcher.exact(self.metric_name),
-            "value_type": LabelMatcher.exact(self.value_type),
+        result: dict[str, str] = {
+            "container_metric_name": self.metric_name,
+            "value_type": self.value_type,
         }
         if self.kernel_id is not None:
-            result["kernel_id"] = LabelMatcher.exact(str(self.kernel_id))
+            result["kernel_id"] = str(self.kernel_id)
         if self.session_id is not None:
-            result["session_id"] = LabelMatcher.exact(str(self.session_id))
+            result["session_id"] = str(self.session_id)
         if self.agent_id is not None:
-            result["agent_id"] = LabelMatcher.exact(self.agent_id)
+            result["agent_id"] = self.agent_id
         if self.user_id is not None:
-            result["user_id"] = LabelMatcher.exact(str(self.user_id))
+            result["user_id"] = str(self.user_id)
         if self.project_id is not None:
-            result["project_id"] = LabelMatcher.exact(str(self.project_id))
+            result["project_id"] = str(self.project_id)
         return result
 
     def group_by_labels(self) -> frozenset[str]:

src/ai/backend/common/clients/valkey_client/valkey_schedule/client.py

@@ -95,14 +95,11 @@ class RouteHealthRecord:
 
     route_id: str
     created_at: int  # Unix timestamp when route was created
-    initial_delay_until: int  # Unix timestamp = running_at + initial_delay
+    initial_delay_until: int  # Unix timestamp = created_at + initial_delay
     health_path: str  # extracted from model_definition
     inference_port: int  # extracted from kernel
     replica_host: str  # extracted from kernel
 
-    # Timestamp when route entered RUNNING state (set by coordinator)
-    running_at: int | None = None
-
     # Agent check results
     agent_healthy: bool = False
     agent_last_check: int = 0  # Unix timestamp
@@ -131,7 +128,7 @@ def is_stale(self, current_time: int, staleness_sec: int = MAX_HEALTH_STALENESS_
 
     def to_valkey_hash(self) -> Mapping[str, str]:
         """Serialize to Valkey hash fields."""
-        data: dict[str, str] = {
+        return {
             "route_id": self.route_id,
             "created_at": str(self.created_at),
             "initial_delay_until": str(self.initial_delay_until),
@@ -143,9 +140,6 @@ def to_valkey_hash(self) -> Mapping[str, str]:
             "manager_healthy": "1" if self.manager_healthy else "0",
             "manager_last_check": str(self.manager_last_check),
         }
-        if self.running_at is not None:
-            data["running_at"] = str(self.running_at)
-        return data
 
     @classmethod
     def from_valkey_hash(cls, data: Mapping[str, str]) -> RouteHealthRecord:
@@ -157,7 +151,6 @@ def from_valkey_hash(cls, data: Mapping[str, str]) -> RouteHealthRecord:
             health_path=data["health_path"],
             inference_port=int(data["inference_port"]),
             replica_host=data["replica_host"],
-            running_at=int(raw) if (raw := data.get("running_at")) and raw != "0" else None,
             agent_healthy=data.get("agent_healthy", "0") == "1",
             agent_last_check=int(data.get("agent_last_check", "0")),
             manager_healthy=data.get("manager_healthy", "0") == "1",
@@ -675,52 +668,6 @@ async def update_route_liveness(self, route_id: str, liveness: bool) -> None:
         async with self._client.client() as conn:
             await conn.exec(batch, raise_on_error=True)
 
-    @valkey_schedule_resilience.apply()
-    async def mark_route_running_at(self, route_id: str) -> None:
-        """
-        Record the RUNNING transition timestamp for a route.
-        Called when a route transitions to RUNNING status.
-        Uses Redis time for consistency with health check comparisons.
-
-        :param route_id: The route ID that entered RUNNING state
-        """
-        key = self._get_route_health_key(route_id)
-        current_time = str(await self._get_redis_time())
-        async with self._client.client() as conn:
-            await conn.hset(key, {"running_at": current_time})
-            await conn.expire(key, ROUTE_HEALTH_TTL_SEC)
-
-    @valkey_schedule_resilience.apply()
-    async def get_route_running_at_batch(self, route_ids: Sequence[str]) -> dict[str, int | None]:
-        """
-        Batch read running_at field from route health hashes.
-        Works even on partial hashes (before full RouteHealthRecord is initialized).
-
-        :param route_ids: Route IDs to look up
-        :return: Mapping of route_id to running_at timestamp (None if not set)
-        """
-        if not route_ids:
-            return {}
-
-        batch = Batch(is_atomic=False)
-        for route_id in route_ids:
-            key = self._get_route_health_key(route_id)
-            batch.hget(key, "running_at")
-
-        async with self._client.client() as conn:
-            results = await conn.exec(batch, raise_on_error=False)
-        if results is None:
-            return dict.fromkeys(route_ids)
-
-        running_at_map: dict[str, int | None] = {}
-        for i, route_id in enumerate(route_ids):
-            raw = results[i] if len(results) > i else None
-            if raw and raw != b"0":
-                running_at_map[route_id] = int(raw)
-            else:
-                running_at_map[route_id] = None
-        return running_at_map
-
     @valkey_schedule_resilience.apply()
     async def refresh_route_health_ttl(self, route_id: str) -> None:
         """
@@ -881,8 +828,6 @@ async def get_route_health_record(self, route_id: str) -> RouteHealthRecord | No
             return None
 
         data = {k.decode(): v.decode() for k, v in result.items()}
-        if "route_id" not in data:
-            return None
         return RouteHealthRecord.from_valkey_hash(data)
 
     @valkey_schedule_resilience.apply()
@@ -921,10 +866,6 @@ async def get_route_health_records_batch(
                 continue
 
             data = {k.decode(): v.decode() for k, v in raw.items()}
-            if "route_id" not in data:
-                # Partial hash (e.g., only running_at set by mark_route_running_at)
-                records[route_id] = None
-                continue
             records[route_id] = RouteHealthRecord.from_valkey_hash(data)
 
         return records