refactor(BA-5713): address slice E review feedback

- repositories/scheduler/types/session.py: rename
  PendingSessionData.access_key -> main_access_key (and drop the
  outdated resolved-main_access_key comment).
- repositories/scheduler/db_source/db_source.py: update the
  PendingSessionData call site to main_access_key + owner_id keyword
  names matching the dataclass.
- scheduler/drf.py: use existing_sess.user_uuid (SessionRow stores the
  owner UUID there, not owner_id).
- scheduler/predicates.py: guard every _resolve_main_access_key
  consumer (check_concurrency, check_keypair_resource_limit,
  check_pending_session_count_limit, check_pending_session_resource_limit)
  with an early main_ak-is-None return so that NULL main_access_key
  users don't fall through to keypair policy lookups that match with
  NULL.

Author: jopemachine

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

@@ -351,9 +351,9 @@ async def _fetch_pending_sessions(
             if session_id not in sessions_map:
                 sessions_map[session_id] = PendingSessionData(
                     id=session_id,
-                    access_key=row.access_key,
+                    main_access_key=row.access_key,
                     requested_slots=row.requested_slots,
-                    user_uuid=row.user_uuid,
+                    owner_id=row.user_uuid,
                     group_id=row.group_id,
                     domain_name=row.domain_name,
                     scaling_group_name=row.scaling_group_name,

src/ai/backend/manager/repositories/scheduler/types/session.py

@@ -44,8 +44,7 @@ class PendingSessionData:
     """Pending session data for scheduling."""
 
     id: SessionId
-    # Resolved main_access_key of the owner; required for keypair-scoped concurrency tracking and resource policy lookups.
-    access_key: AccessKey
+    main_access_key: AccessKey
     requested_slots: ResourceSlot
     owner_id: UUID
     group_id: UUID
@@ -65,7 +64,7 @@ def to_session_workload(self) -> SessionWorkload:
         kernel_workloads = [k.to_kernel_workload() for k in self.kernels]
         return SessionWorkload(
             session_id=self.id,
-            main_access_key=self.access_key,
+            main_access_key=self.main_access_key,
             requested_slots=self.requested_slots,
             owner_id=self.owner_id,
             group_id=self.group_id,

src/ai/backend/manager/scheduler/drf.py

@@ -60,7 +60,7 @@ def pick_session(
                 slot_share = Decimal(value) / slot_cap
                 if dominant_share < slot_share:
                     dominant_share = slot_share
-            owner_id = existing_sess.owner_id
+            owner_id = existing_sess.user_uuid
             if owner_id is not None:
                 if self.per_user_dominant_share[owner_id] < dominant_share:
                     self.per_user_dominant_share[owner_id] = dominant_share

src/ai/backend/manager/scheduler/predicates.py

@@ -60,6 +60,11 @@ async def check_concurrency(
     sess_ctx: SessionRow,
 ) -> PredicateResult:
     main_ak = await _resolve_main_access_key(db_sess, sess_ctx)
+    if main_ak is None:
+        return PredicateResult(
+            False,
+            "Session owner has no main_access_key; cannot evaluate concurrency policy",
+        )
 
     async def _get_max_concurrent_sessions() -> int:
         resouce_policy_q = sa.select(KeyPairRow.resource_policy).where(
@@ -144,6 +149,8 @@ async def check_keypair_resource_limit(
     sess_ctx: SessionRow,
 ) -> PredicateResult:
     main_ak = await _resolve_main_access_key(db_sess, sess_ctx)
+    if main_ak is None:
+        return PredicateResult(False, "Session owner has no main_access_key")
     resouce_policy_q = sa.select(KeyPairRow.resource_policy).where(KeyPairRow.access_key == main_ak)
     select_query = sa.select(KeyPairResourcePolicyRow).where(
         KeyPairResourcePolicyRow.name == resouce_policy_q.scalar_subquery()
@@ -162,9 +169,6 @@ async def check_keypair_resource_limit(
     total_keypair_allowed = ResourceSlot.from_policy(
         resource_policy_map, cast(Mapping[str, Any], sched_ctx.known_slot_types)
     )
-
-    if main_ak is None:
-        return PredicateResult(False, "Session has no access key")
     key_occupied = await sched_ctx.registry.get_keypair_occupancy(
         AccessKey(main_ak), db_sess=db_sess
     )
@@ -308,6 +312,8 @@ async def check_pending_session_count_limit(
     failure_msgs = []
 
     main_ak = await _resolve_main_access_key(db_sess, sess_ctx)
+    if main_ak is None:
+        return PredicateResult(False, "Session owner has no main_access_key")
     query = (
         sa.select(SessionRow)
         .where(
@@ -370,6 +376,8 @@ async def check_pending_session_resource_limit(
     failure_msgs = []
 
     main_ak = await _resolve_main_access_key(db_sess, sess_ctx)
+    if main_ak is None:
+        return PredicateResult(False, "Session owner has no main_access_key")
     query = (
         sa.select(SessionRow)
         .where(