refactor(BA-5714): resolve rebase conflicts and add resolve_main_access_keys

- Resolve merge conflicts from cascading onto the updated slice E
  (drf.py keeps ``existing_sess.user_uuid``; predicates.py keeps the
  ``main_ak is None`` early returns; scheduler types stay on
  ``main_access_key``; stream db_source keeps ``UserNotFound``).
- Add ``SchedulerRepository.resolve_main_access_keys`` and the matching
  ``ScheduleDBSource.resolve_main_access_keys`` so the coordinator's
  cache-invalidation step can look up each session's owner
  main_access_key in a single query. Required by
  sokovan/scheduler/coordinator.py and lifecycle/deprioritize_sessions.py
  call sites that were previously reading ``session_info.metadata.access_key``.

Author: jopemachine

changes/BA-5650-E.misc.md

@@ -4774,6 +4774,26 @@ async def get_db_now(self) -> datetime:
             result = await conn.execute(sa.select(sa.func.now()))
             return result.scalar_one()
 
+    async def resolve_main_access_keys(
+        self, session_ids: Sequence[SessionId]
+    ) -> dict[SessionId, AccessKey]:
+        """Resolve the main access key for each session's owner.
+
+        Joins ``sessions`` → ``users`` to look up the owner's
+        ``main_access_key``. Sessions whose owner has no configured
+        main access key are omitted from the returned mapping.
+        """
+        if not session_ids:
+            return {}
+        async with self._db.begin_readonly_session() as db_sess:
+            stmt = (
+                sa.select(SessionRow.id, UserRow.main_access_key)
+                .join(UserRow, SessionRow.user_uuid == UserRow.uuid)
+                .where(SessionRow.id.in_([sid for sid in session_ids]))
+            )
+            rows = (await db_sess.execute(stmt)).all()
+            return {SessionId(row[0]): AccessKey(row[1]) for row in rows if row[1] is not None}
+
     async def _get_db_now_in_session(self, db_sess: SASession) -> datetime:
         """Get the current timestamp from the database within an existing session.
 

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

@@ -959,3 +959,9 @@ async def get_db_now(self) -> datetime:
             Current database timestamp with timezone
         """
         return await self._db_source.get_db_now()
+
+    async def resolve_main_access_keys(
+        self, session_ids: Sequence[SessionId]
+    ) -> dict[SessionId, AccessKey]:
+        """Resolve the main access key for each session's owner."""
+        return await self._db_source.resolve_main_access_keys(session_ids)